本文介紹了Langgraph服務的四種調用方式：

1. 通過LangGraph Studio UI界面手動測試；

2. 使用Python SDK進行同步/異步調用；

3. 通過REST API測試；

4. 使用JavaScript SDK接入。

Langgraph 服務端代碼?graph.py

from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agentllm = ChatOpenAI(model='qwq-32b',temperature=0.8,api_key='sk-****',streaming=True,base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",# extra_body={'chat_template_kwargs': {'enable_thinking': False}},
)
#
def get_weather(city: str) -> str:"""Get weather for a given city."""return f"在 {city}，今天天氣不錯!"graph = create_react_agent(llm,tools=[get_weather],prompt="你是一個智能助手"
)

通過命令langgraph dev 啟動服務，可以看到控制臺返回的API地址

1.第一種訪問方式：LangGraph Studio

當啟動服務后，瀏覽器中會自動打開 Studio UI的地址，頁面如下

可手動輸入message，完成調用

2.第二種訪問方式：PythonSDK測試

先安裝依賴：pip install langgraph-sdk

1）異步測試

langgraph_async_test.py文件內容：

from langgraph_sdk import get_client
import asyncioclient = get_client(url="http://localhost:2024")async def main():async for chunk in client.runs.stream(None,  # Threadless run"agent", # Name of assistant. Defined in langgraph.json.input={"messages": [{"role": "human","content": "上海今天的天氣",}],},stream_mode="messages-tuple",):# print(f"Receiving new event of type: {chunk.event}...")# print(chunk.data)if isinstance(chunk.data,list) and 'type' in chunk.data[0] and chunk.data[0]['type'] == 'AIMessageChunk':print(chunk.data[0]['content'], end='', flush=True)asyncio.run(main())

運行結果：

2）同步測試

langgraph_sync_test.py文件內容：

from langgraph_sdk import  get_sync_clientclient = get_sync_client(url="http://localhost:2024")for chunk in client.runs.stream(None,  # Threadless run"agent", # Name of assistant. Defined in langgraph.json.input={"messages": [{"role": "human","content": "上海今天的天氣",}],},stream_mode="messages-tuple",):# print(f"Receiving new event of type: {chunk.event}...")# print(chunk.data)if isinstance(chunk.data,list) and 'type' in chunk.data[0] and chunk.data[0]['type'] == 'AIMessageChunk':print(chunk.data[0]['content'], end='', flush=True)

運行結果：

后面2種方式，可自動測試，參考如下

3.第三種訪問方式：REST API測試

curl -s --request POST \--url "http://localhost:2024/runs/stream" \--header 'Content-Type: application/json' \--data "{\"assistant_id\": \"agent\",\"input\": {\"messages\": [{\"role\": \"human\",\"content\": \"上海的天氣?\"}]},\"stream_mode\": \"messages-tuple\"}"

4.第四種訪問方式：JavaScript SDK測試

安裝 LangGraph JS SDK：npm install @langchain/langgraph-sdk

向LangGraph服務區發送消息：

const { Client } = await import("@langchain/langgraph-sdk");// only set the apiUrl if you changed the default port when calling langgraph dev
const client = new Client({ apiUrl: "http://localhost:2024"});const streamResponse = client.runs.stream(null, // Threadless run"agent", // Assistant ID{input: {"messages": [{ "role": "user", "content": "上海的天氣?"}]},streamMode: "messages-tuple",}
);for await (const chunk of streamResponse) {console.log(`Receiving new event of type: ${chunk.event}...`);console.log(JSON.stringify(chunk.data));console.log("\n\n");
}

graph.py中的get_weather方法可替找成工具

TAVILY_API_KEY='tvly-dev-***'def get_weather(city: str) -> str:"""Get real-time weather for a given city using Tavily search."""from langchain_community.tools import TavilySearchResultsimport re# 創建 Tavily 搜索工具實例search_tool = TavilySearchResults(max_results=1,search_depth="advanced",include_answer=True,api_key=TAVILY_API_KEY)# 構造搜索查詢query = f"{city} 當前天氣 溫度 濕度 風力 風向 天氣狀況"try:# 執行搜索results = search_tool.invoke({"query": query})# 解析結果if isinstance(results, list) and len(results) > 0:result = results[0]content = result.get('content', '暫無詳細信息')# 修復編碼問題的函數def fix_encoding(text):if not isinstance(text, str):return text# 嘗試多種編碼修復方法encodings_to_try = [('latin1', 'utf-8'),('cp1252', 'utf-8'),]for from_enc, to_enc in encodings_to_try:try:# 將字符串以from_enc編碼方式重新編碼，再以to_enc解碼return text.encode(from_enc).decode(to_enc)except (UnicodeEncodeError, UnicodeDecodeError):continue# 如果上面的方法都不行，嘗試直接使用raw_unicode_escapetry:return text.encode('raw_unicode_escape').decode('utf-8')except (UnicodeEncodeError, UnicodeDecodeError):pass# 如果所有方法都失敗，返回原始內容return text# 修復編碼fixed_content = fix_encoding(content)print(f"處理后內容: {fixed_content}")# 從修復后的內容中提取天氣信息def extract_weather_info(text):info = {}# 提取天氣狀況（如晴、多云、陰等）weather_conditions = ['晴', '多云', '陰', '雨', '雪', '霧', '霾', '雷陣雨', '小雨', '中雨', '大雨', '暴雨']for condition in weather_conditions:if condition in text:info['condition'] = conditionbreak# 如果沒找到中文天氣狀況，嘗試用正則表達式if 'condition' not in info:condition_match = re.search(r'天氣[:\s]*([^\s,，。\.]+)', text)if condition_match:info['condition'] = condition_match.group(1)# 提取溫度 (尋找數字+度/℃)temp_match = re.search(r'(\d+\.?\d*)[度℃]', text)if temp_match:info['temperature'] = temp_match.group(1) + "℃"# 提取濕度humidity_match = re.search(r'濕度[:\s]*([0-9]+\.?[0-9]*)[%％]', text)if humidity_match:info['humidity'] = humidity_match.group(1) + "%"# 提取風向wind_dir_match = re.search(r'風向[:\s]*([東南西北風]+)', text)if wind_dir_match:info['wind_direction'] = wind_dir_match.group(1)# 提取風力wind_speed_match = re.search(r'風力[:\s]*([0-9]+\.?[0-9]*[級m/s])', text)if wind_speed_match:info['wind_speed'] = wind_speed_match.group(1)return info# 提取天氣信息weather_info = extract_weather_info(fixed_content)# 構建最終輸出格式if weather_info:output_parts = [f"{city} 天氣"]if 'condition' in weather_info:output_parts.append(f"{weather_info['condition']}")output_parts.append("實時天氣情況")if 'temperature' in weather_info:output_parts.append(f"溫度{weather_info['temperature']}")if 'humidity' in weather_info:output_parts.append(f"濕度{weather_info['humidity']}")if 'wind_direction' in weather_info:output_parts.append(f"風向{weather_info['wind_direction']}")if 'wind_speed' in weather_info:output_parts.append(f"風力{weather_info['wind_speed']}")return " ".join(output_parts)else:# 如果無法提取結構化信息，則返回修復后的內容return f"{city}天氣信息: {fixed_content}"else:return f"無法獲取{city}的天氣信息"except Exception as e:# 異常處理return f"查詢{city}天氣時出現錯誤: {str(e)}"

替換后再次測試，結果如下：