使用LangGraph构建智能代理#

本示例演示了如何使用LangGraph创建一个AI代理。基于LangGraph文档中的示例： https://langchain-ai.github.io/langgraph/.

首先安装依赖项：

# pip install langgraph langchain-openai azure-identity

导入所需模块。

from dataclasses import dataclass
from typing import Any, Callable, List, Literal

from autogen_core import AgentId, MessageContext, RoutedAgent, SingleThreadedAgentRuntime, message_handler
from azure.identity import DefaultAzureCredential, get_bearer_token_provider
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_core.tools import tool  # pyright: ignore
from langchain_openai import AzureChatOpenAI, ChatOpenAI
from langgraph.graph import END, MessagesState, StateGraph
from langgraph.prebuilt import ToolNode

定义用于与代理通信的消息类型。

@dataclass
class Message:
    content: str

定义代理将使用的工具。

@tool  # pyright: ignore
def get_weather(location: str) -> str:
    """Call to surf the web."""
    # 这是一个占位符，但不要告诉大语言模型...
    if "sf" in location.lower() or "san francisco" in location.lower():
        return "It's 60 degrees and foggy."
    return "It's 90 degrees and sunny."

使用LangGraph的API定义代理。

class LangGraphToolUseAgent(RoutedAgent):
    def __init__(self, description: str, model: ChatOpenAI, tools: List[Callable[..., Any]]) -> None:  # pyright: ignore
        super().__init__(description)
        self._model = model.bind_tools(tools)  # pyright: ignore

        # 定义判断是否继续执行的函数
        def should_continue(state: MessagesState) -> Literal["tools", END]:  # type: ignore
            messages = state["messages"]
            last_message = messages[-1]
            # 如果LLM发起工具调用，则路由到"tools"节点
            if last_message.tool_calls:  # type: ignore
                return "tools"
            # 否则，我们停止（回复用户）
            return END

        # 定义调用模型的函数
        async def call_model(state: MessagesState):  # type: ignore
            messages = state["messages"]
            response = await self._model.ainvoke(messages)
            # 我们返回一个列表，因为这将被添加到现有列表中
            return {"messages": [response]}

        tool_node = ToolNode(tools)  # pyright: ignore

        # 定义一个新图
        self._workflow = StateGraph(MessagesState)

        # 定义我们将循环的两个节点
        self._workflow.add_node("agent", call_model)  # pyright: ignore
        self._workflow.add_node("tools", tool_node)  # pyright: ignore

        # 将入口点设置为`agent`，这意味着该节点是第一个被调用的
        # This means that this node is the first one called
        self._workflow.set_entry_point("agent")

        # 现在添加一个条件边
        self._workflow.add_conditional_edges(
            # 首先，我们定义起始节点。使用 `agent` 节点，
            # 这意味着这些边是在调用 `agent` 节点后采取的路径。
            "agent",
            # 接着，我们传入一个函数来决定下一个调用的节点。
            should_continue,  # type: ignore
        )

        # 现在添加从 `tools` 到 `agent` 的普通边，
        # 这意味着在调用 `tools` 后， 下一个将调用 `agent` 节点。
        self._workflow.add_edge("tools", "agent")

        # 最后，
        # 我们进行编译！这将把它编译成一个LangChain可运行对象，
        # 意味着你可以像使用其他可运行对象一样使用它。
        # 注意我们在编译图时（可选地）传递了内存
        self._app = self._workflow.compile()

    @message_handler
    async def handle_user_message(self, message: Message, ctx: MessageContext) -> Message:
        # 使用可运行对象
        final_state = await self._app.ainvoke(
            {
                "messages": [
                    SystemMessage(
                        content="You are a helpful AI assistant. You can use tools to help answer questions."
                    ),
                    HumanMessage(content=message.content),
                ]
            },
            config={"configurable": {"thread_id": 42}},
        )
        response = Message(content=final_state["messages"][-1].content)
        return response

现在让我们测试这个代理。首先需要创建一个代理运行时环境并注册代理，通过提供代理名称和一个用于创建代理的工厂函数。

runtime = SingleThreadedAgentRuntime()
await LangGraphToolUseAgent.register(
    runtime,
    "langgraph_tool_use_agent",
    lambda: LangGraphToolUseAgent(
        "Tool use agent",
        ChatOpenAI(
            model="gpt-4o",
            # api_key=os.getenv("OPENAI_API_KEY"),
        ),
        # AzureChatOpenAI( azure_deployment=os.getenv("AZURE_OPENAI_DEPLOYMENT"),
        # azure_endpoint=os.getenv("AZURE_OPENAI_ENDPOINT"),
        # api_version=os.getenv("AZURE_OPENAI_API_VERSION"),
        # # 使用Azure Active Directory认证 azure_ad_token_provider=get_bearer_token_provider(DefaultAzureCredential()),
        # # 使用API密钥 # api_key=os.getenv("AZURE_OPENAI_API_KEY"),
        # ),
        # 
        # 
        # 
        [get_weather],
    ),
)
agent = AgentId("langgraph_tool_use_agent", key="default")

启动代理运行时。

runtime.start()

向代理发送直接消息，并打印响应。

response = await runtime.send_message(Message("What's the weather in SF?"), agent)
print(response.content)

The current weather in San Francisco is 60 degrees and foggy.

停止代理运行时。

await runtime.stop()