大家都知道langchain,LangChain是一个用于开发由语言模型驱动的应用程序的框架。它提供了一套工具和组件,帮助开发者更轻松地构建复杂的LLM应用。

LangGraph是LangChain生态系统中的一个库,专门用于构建有状态、多参与者(Agent)的应用程序。它基于图(Graph)的概念,可以创建复杂的、循环的工作流。

废话不多说,模拟实现langgraph的create_react_agent

1. 核心 State 设计

from typing import TypedDict, List
from langchain_core.messages import BaseMessage

class AgentState(TypedDict):
    messages: List[BaseMessage]

2. 自定义工具

@tool
def get_weather(city: str) -> str:
    """根据城市获取天气信息"""
    return f"{city}天气是晴天 25度"


@tool
def get_time(city: str) -> str:
    """根据城市获取现在时间"""
    return f"{city}时间2026年3月23日"

3. 初始化模型(需要支持tool calling)

import os
import dotenv
from langchain_openai import ChatOpenAI

dotenv.load_dotenv()

os.environ["OPENAI_API_KEY"] = os.getenv("ALY_API_KEY")
os.environ["OPENAI_BASE_URL"] = os.getenv("ALY_BASE_URL")
model_name = os.getenv("MODEL_NAME")

llm = ChatOpenAI(
    model=model_name,
    temperature=0
).bind_tools([get_weather, get_time])

4. 定义节点

from langchain_core.messages import ToolMessage

def llm_node(state):
    messages = state["messages"]
    response = llm.invoke(messages)
    return {"messages": messages + [response]}

def tool_node(state):
    last_msg = state["messages"][-1]

    tool_calls = last_msg.tool_calls

    results = []
    for call in tool_calls:
        tool_name = call["name"]
        args = call["args"]

        if tool_name == "get_weather":
            result = get_weather.invoke(args)

        if tool_name == "get_time":
            result = get_time.invoke(args)

        results.append(
            ToolMessage(
                content=result,
                tool_call_id=call["id"]
            )
        )

    return {"messages": state["messages"] + results}

def should_continue(state):
    last_msg = state["messages"][-1]
    if last_msg.tool_calls:
        return "tool"
    return "end"

5. 构建LangGraph

from langgraph.graph import StateGraph, END

graph = StateGraph(AgentState)

# 添加节点到graph
graph.add_node("llm", llm_node)
graph.add_node("tool", tool_node)

# 设置入口,等同于graph.add_edge(START, "llm")
graph.set_entry_point("llm")

# 添加条件判断边
graph.add_conditional_edges(
    "llm",
    should_continue,
    {
        "tool": "tool",
        "end": END
    }
)

# 设置节点运转的边
graph.add_edge("tool", "llm")

# 编译图
app = graph.compile()

6. 最后让我们来试着调用一下模型提问吧

from langchain_core.messages import HumanMessage

result = app.invoke(
    input={
        "messages": [
            HumanMessage(content="北京的天气怎么样?北京昨天日期是多少?")
        ]
    })

print(result)

7. 通过print result观察模型调用的日志

8. 最后让我们看一看我们定义的这个图长什么样吧(需要用Jupyter)

from IPython.display import Image
from test_v2_02.a import app  # 这个是你原本.py文件里的graph变量

Image(app.get_graph().draw_mermaid_png())

Logo

Agent 垂直技术社区,欢迎活跃、内容共建。

更多推荐