如何把 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.mdPython 3.10 或更高版本是autogen-core的要求。安装依赖pip install autogen-core pip install langgraph langchain-openai azure-identity第二条命令来自 LangGraph 示例文档开头。其中azure-identity仅在使用示例中被注释掉的 Azure OpenAI 分支时需要该分支用 AAD 令牌认证。示例中模型客户端为ChatOpenAI(modelgpt-4o)。注意示例里api_keyos.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 dont tell the LLM that... if sf in location.lower() or san francisco in location.lower(): return Its 60 degrees and foggy. return Its 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 were (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( contentYou are a helpful AI assistant. You can use tools to help answer questions. ), HumanMessage(contentmessage.content), ] }, config{configurable: {thread_id: 42}}, ) response Message(contentfinal_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( modelgpt-4o, # api_keyos.getenv(OPENAI_API_KEY), ), [get_weather], ), ) agent AgentId(langgraph_tool_use_agent, keydefault)AgentId由类型对应工厂注册的名称和 key 组成首次向该AgentId投递消息时运行时会按工厂创建实例见 Agent and Agent Runtime 中“Registering Agent Type”一节。启动运行时并验证消息收发runtime.start() response await runtime.send_message(Message(Whats 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_providerget_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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考