iT邦幫忙

2025 iThome 鐵人賽

DAY 6
0
生成式 AI

nutc_imac_Agent拼裝車系列 第 6

Day 06 實戰 (一)

  • 分享至 

  • xImage
  •  

一、程式碼範例(Python)

from typing import TypedDict
from langgraph.graph import StateGraph, END, START
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage, ToolMessage
from langchain_core.tools import tool
from pydantic import BaseModel, Field

import os
import requests

# 定義 state 的結構
class MyAgentState(TypedDict):
    messages: list[BaseMessage]  # 用來儲存對話訊息
    counter: int

# 定義工具(tool)
class WeatherInput(BaseModel):
    location: str = Field(..., description="城市名稱,例如 Taipei")
    date: str = Field(..., description="查詢日期,格式 YYYY-MM-DD")

@tool("get_weather", args_schema=WeatherInput, return_direct=True)
def get_weather(location: str, date: str):
    # 簡單示意,實際要換成真的天氣 API
    response = requests.get(
        f"https://api.example.com/weather?loc={location}&date={date}"
    )
    if response.status_code == 200:
        return {"weather": response.json()}
    else:
        return {"error": f"Status code {response.status_code}"}

# 定義 node 函式
def call_model_node(state: MyAgentState):
    """
    用 LLM 處理最新對話,看是否要呼叫工具
    假設 state["messages"] 最後一條訊息可能包含 tool_calls
    """
    # 這裡假設已有一個 model_with_tools
    # model_with_tools.invoke(...) 回傳 BaseMessage
    # 為了示意,我這裡直接回傳一個 ToolMessage 或普通訊息
    
    # 假設 last message 想查天氣
    tool_msg = ToolMessage(
        content=None,
        name="get_weather",
        tool_call_id="call1",
        # args 跟 tool schema 相符
        **{"args": {"location": "Taipei", "date": "2025-09-21"}}
    )
    return {"messages": [tool_msg]}

def call_tool_node(state: MyAgentState):
    """
    處理 tool 呼叫
    """
    outputs = []
    last_msg = state["messages"][-1]
    if isinstance(last_msg, ToolMessage):
        # 執行工具
        res = get_weather.invoke(**last_msg.tool_call_args)  # 假設有這屬性
        outputs.append(
            ToolMessage(
                content=res,
                name=last_msg.name,
                tool_call_id=last_msg.tool_call_id,
            )
        )
    return {"messages": outputs, "counter": state["counter"] + 1}

def decide_next(state: MyAgentState):
    """決定流程是否結束或繼續呼叫 model 或 tool"""
    last_msg = state["messages"][-1]
    # 如果最後訊息不是 ToolMessage,就結束
    if not isinstance(last_msg, ToolMessage):
        return END
    # 否則回到 model node 繼續處理
    return "call_model"

# 建構 graph
workflow = StateGraph(state_schema=MyAgentState)
workflow.add_node("call_model", call_model_node)
workflow.add_node("call_tool", call_tool_node)
workflow.set_entry_point("call_model")
workflow.add_edge("call_model", "call_tool")
workflow.add_conditional_edges(source="call_tool", path=decide_next)

# 初始化 state
initial_state: MyAgentState = {
    "messages": [],
    "counter": 0,
}

# 執行 graph
graph = workflow.compile()
final_state = graph.invoke(initial_state)
print(final_state)

二、說明

名稱 用途
StateGraph 定義整個工作流程的圖(graph),包含狀態 (state)、節點 (nodes)、邊 (edges)
state_schema 定義狀態裡會有哪些欄位 (fields),也規範類型
@tool 定義一個工具(tool),可以在流程中被節點呼叫
node 函式 每個 node 接收目前的 state,回傳該 node 處理後的部分 state 變更
edge 普通邊 (無條件) 和條件邊 (conditional edge);條件邊可以依 state 決定接下來要去哪個 node 或是結束流程
END / START LangGraph 提供的特殊標記,用來指示流程結束或開始的地方

三、資料來源

  • LangGraph 官方文件:「Learn LangGraph basics」概念與教學介紹 (langchain-ai.github.io)
  • LangChain 官方網站關於 LangGraph 架構與特性說明 (langchain.com)
  • LangChain Academy 課程:Introduction to LangGraph (LangChain Academy)
  • Futuresmart Blog 教學文章:《LangGraph Tutorial: A Comprehensive Guide for Beginners》 (blog.futuresmart.ai)

上一篇
Day 05 當前 AI Agent 設計框架介紹
下一篇
Day 07 實戰 (二)
系列文
nutc_imac_Agent拼裝車8
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言