企业官网建设流程全解析

如何把 LangGraph 智能体接入 AutoGen Core 运行时？

【免费下载链接】autogenA programming framework for agentic AI项目地址: https://gitcode.com/GitHub_Trending/au/autogen

如果你已经用 LangGraph 搭好了一个带工具调用（tool calling）的工作流，希望把它作为一个智能体放进 AutoGen 的 Agent 运行时里，由运行时统一管理生命周期并通过消息收发与它通信，AutoGen 官方在 Core 文档的 Cookbook 中给出了完整示例：langgraph-agent.ipynb。

接入方式的核心是：把 LangGraph 的StateGraph封装进一个继承自RoutedAgent的 AutoGen 智能体类，在消息处理器中调用编译后的 LangGraph Runnable，然后把这个智能体注册到SingleThreadedAgentRuntime（本地嵌入式运行时）即可。整个过程不依赖 LangGraph 之外的其他 AutoGen 高层封装，直接面向 Core API。

准备条件

根据 installation.md，Python 3.10 或更高版本是autogen-core的要求。安装依赖：

pip install "autogen-core" pip install langgraph langchain-openai azure-identity

第二条命令来自 LangGraph 示例文档开头。其中azure-identity仅在使用示例中被注释掉的 Azure OpenAI 分支时需要（该分支用 AAD 令牌认证）。

示例中模型客户端为ChatOpenAI(model="gpt-4o")。注意示例里api_key=os.getenv("OPENAI_API_KEY")这一行在 notebook 中是注释状态，照抄示例时需要取消注释或改为自己的鉴权方式，否则模型调用不会带上密钥。

定义消息类型和工具

AutoGen Core 中智能体之间通过消息类型通信。示例先定义一个 dataclass 作为与智能体通信的消息类型：

@dataclass class Message: content: str

再定义 LangGraph 工作流要使用的工具（示例中的get_weather是一个占位实现）：

@tool # pyright: ignore def get_weather(location: str) -> str: """Call to surf the web.""" # This is a placeholder, but don't tell the LLM that... 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."

用 RoutedAgent 封装 LangGraph 工作流

AutoGen Core 的智能体通常继承 {py:class}~autogen_core.RoutedAgent，用@message_handler装饰器声明每种消息类型的处理方法（见 Agent and Agent Runtime）。下面是示例中封装 LangGraph 的完整智能体类：

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 # Define the function that determines whether to continue or not def should_continue(state: MessagesState) -> Literal["tools", END]: # type: ignore messages = state["messages"] last_message = messages[-1] # If the LLM makes a tool call, then we route to the "tools" node if last_message.tool_calls: # type: ignore return "tools" # Otherwise, we stop (reply to the user) return END # Define the function that calls the model async def call_model(state: MessagesState): # type: ignore messages = state["messages"] response = await self._model.ainvoke(messages) # We return a list, because this will get added to the existing list return {"messages": [response]} tool_node = ToolNode(tools) # pyright: ignore # Define a new graph self._workflow = StateGraph(MessagesState) # Define the two nodes we will cycle between self._workflow.add_node("agent", call_model) # pyright: ignore self._workflow.add_node("tools", tool_node) # pyright: ignore # Set the entrypoint as `agent` # This means that this node is the first one called self._workflow.set_entry_point("agent") # We now add a conditional edge self._workflow.add_conditional_edges( # First, we define the start node. We use `agent`. # This means these are the edges taken after the `agent` node is called. "agent", # Next, we pass in the function that will determine which node is called next. should_continue, # type: ignore ) # We now add a normal edge from `tools` to `agent`. # This means that after `tools` is called, `agent` node is called next. self._workflow.add_edge("tools", "agent") # Finally, we compile it! # This compiles it into a LangChain Runnable, # meaning you can use it as you would any other runnable. # Note that we're (optionally) passing the memory when compiling the graph self._app = self._workflow.compile() @message_handler async def handle_user_message(self, message: Message, ctx: MessageContext) -> Message: # Use the Runnable 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

结构上分为两部分：

• 构造函数内按 LangGraph 的 API 搭建图：call_model节点调用绑定了工具的模型，should_continue决定 LLM 发起工具调用时路由到tools节点、否则结束（END），工具执行完再通过tools -> agent的边回到模型节点，最后compile()得到可运行的self._app。
• handle_user_message是接入 AutoGen Core 的入口：收到Message后把系统提示和用户消息组装起来调用self._app.ainvoke，取最终状态的最后一条消息内容作为Message返回。config中的thread_id: 42是示例 notebook 里的固定值，可按自己的会话标识替换。

导入清单（与示例一致）：

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

注册智能体并接入运行时

运行时负责创建和管理智能体实例：你只需要提供智能体名称和一个创建实例的工厂函数。示例中注册代码为：

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"), ), [get_weather], ), ) agent = AgentId("langgraph_tool_use_agent", key="default")

AgentId由类型（对应工厂注册的名称）和 key 组成；首次向该AgentId投递消息时，运行时会按工厂创建实例（见 Agent and Agent Runtime 中“Registering Agent Type”一节）。

启动运行时并验证消息收发

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

runtime.start()启动后台消息处理任务，await runtime.stop()立即停止。示例 notebook 实际运行后的输出（文档示例）为：

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

这条回复与get_weather对 "sf" 的占位返回值一致，说明消息经 AutoGen Core 进入handle_user_message、由 LangGraph 图走完模型与工具节点后返回。

限制与注意事项

• 脚本形式运行：notebook 中的register、send_message、stop都用了顶层await，只适用于 Jupyter。Quickstart 文档说明：在 VS Code 等编辑器里应导入asyncio，把上述代码包进async def main() -> None:中，并用asyncio.run(main())执行。
• Azure OpenAI 可选分支：示例 notebook 中注释保留了AzureChatOpenAI的写法，通过环境变量AZURE_OPENAI_DEPLOYMENT、AZURE_OPENAI_ENDPOINT、AZURE_OPENAI_API_VERSION配置，并支持azure_ad_token_provider=get_bearer_token_provider(DefaultAzureCredential())（AAD 认证）或直接传api_key。需要时取消注释替换ChatOpenAI，并确认已安装azure-identity。
• 运行时的角色边界：智能体实例由运行时按需创建和管理，应用代码不直接持有实例；AgentId只用于与智能体通信或读取元数据。SingleThreadedAgentRuntime是本地嵌入式运行时；AutoGen Core 另有分布式运行时（可在 Quickstart 和 distributed-agent-runtime.ipynb 中查看），但 LangGraph 示例只演示了单线程本地场景。
• 线程 ID 是示例值：thread_id: 42仅为 notebook 示例写法，示例文档未解释其多租户或会话隔离语义，不要把它当作推荐配置直接复用。

【免费下载链接】autogenA programming framework for agentic AI项目地址: https://gitcode.com/GitHub_Trending/au/autogen