Part 11 · 多 Agent 与 Agent IDE

import osimport asynciofrom dotenv import load_dotenvfrom autogen_agentchat.agents import AssistantAgentfrom autogen_agentchat.teams import RoundRobinGroupChatfrom autogen_agentchat.ui import Consolefrom autogen_ext.models.openai import OpenAIChatCompletionClientload_dotenv()# 模型客户端（DeepSeek 兼容 OpenAI 接口）# deepseek-chat 不在 AutoGen 内置模型白名单，需手动提供 model_infomodel_client = OpenAIChatCompletionClient(    model="deepseek-chat",    api_key=os.getenv("DEEPSEEK_API_KEY"),    base_url="https://api.deepseek.com",    model_info={        "vision": False,        "function_calling": True,        "json_output": True,        "structured_output": True,        "family": "unknown",    },)# 两个角色：一个提问，一个回答#  agent 的 name 只能用英文字母、数字、下划线、连字符，不能用中文questioner = AssistantAgent(    name="questioner",    model_client=model_client,    system_message="你是提问者，负责提出有趣的问题。每次只提一个问题，不要回答。",)answerer = AssistantAgent(    name="answerer",    model_client=model_client,    system_message="你是回答者，负责详细回答提问者的问题。",)# 轮流对话团队，最多 3 轮team = RoundRobinGroupChat([questioner, answerer], max_turns=3)# 新版 AutoGen 是 async 的，必须用 asyncio.runasync def main():    await Console(team.run_stream(task="我们来聊聊人工智能。你先提个问题吧。"))if __name__ == "__main__":    asyncio.run(main())

import osimport asynciofrom dotenv import load_dotenvfrom autogen_agentchat.agents import AssistantAgentfrom autogen_agentchat.teams import RoundRobinGroupChatfrom autogen_agentchat.ui import Consolefrom autogen_ext.models.openai import OpenAIChatCompletionClientload_dotenv()model_client = OpenAIChatCompletionClient(    model="deepseek-chat",    api_key=os.getenv("DEEPSEEK_API_KEY"),    base_url="https://api.deepseek.com",    model_info={"vision": False, "function_calling": True, "json_output": True, "structured_output": True, "family": "unknown"},)planner = AssistantAgent(name="planner", model_client=model_client, system_message="你负责把任务拆成步骤，输出计划后结束。")coder = AssistantAgent(name="coder", model_client=model_client, system_message="你是Python工程师，负责写代码。")critic = AssistantAgent(name="critic", model_client=model_client, system_message="你审查代码，指出问题后结束。")team = RoundRobinGroupChat([planner, coder, critic], max_turns=3)async def main():    await Console(team.run_stream(task="写一个Python函数：输入数字n，返回1到n的和。"))if __name__ == "__main__":    asyncio.run(main())

from typing import TypedDict, Literalfrom langgraph.graph import StateGraph, ENDclass State(TypedDict):    value: int    result: strdef check(state: State):    v = state["value"]    return {"result": "大" if v > 10 else "小"}# 根据条件走不同边def route(state: State) -> Literal["big", "small"]:    return "big" if state["value"] > 10 else "small"def big(state: State):    return {"result": state["result"] + "于10"}def small(state: State):    return {"result": state["result"] + "于等于10"}g = StateGraph(State)g.add_node("check", check)g.add_node("big", big)g.add_node("small", small)g.set_entry_point("check")g.add_conditional_edges("check", route, {"big": "big", "small": "small"})g.add_edge("big", END)g.add_edge("small", END)app = g.compile()print(app.invoke({"value": 15}))   # result: 大于10print(app.invoke({"value": 3}))    # result: 小于等于10

from typing import TypedDictfrom langgraph.graph import StateGraph, ENDfrom langchain_openai import ChatOpenAIfrom langchain_core.tools import toolfrom langchain_core.messages import HumanMessageimport osfrom dotenv import load_dotenvload_dotenv()llm = ChatOpenAI(model="deepseek-chat", api_key=os.getenv("DEEPSEEK_API_KEY"), base_url="https://api.deepseek.com")@tooldef add(a: float, b: float) -> float:    """相加"""    return a + btools = [add]llm_with_tools = llm.bind_tools(tools)class State(TypedDict):    messages: listdef agent_node(state):    resp = llm_with_tools.invoke(state["messages"])    return {"messages": state["messages"] + [resp]}def tools_node(state):    last = state["messages"][-1]    results = []    for call in last.tool_calls:        if call["name"] == "add":            results.append({"role": "tool", "content": str(add.invoke(call["args"])), "tool_call_id": call["id"]})    return {"messages": state["messages"] + results}def should_continue(state):    last = state["messages"][-1]    return "tools" if getattr(last, "tool_calls", None) else ENDg = StateGraph(State)g.add_node("agent", agent_node)g.add_node("tools", tools_node)g.set_entry_point("agent")g.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END})g.add_edge("tools", "agent")app = g.compile()result = app.invoke({"messages": [HumanMessage("计算 7 加 8 等于多少")]})print("最终回答:", result["messages"][-1].content)

from langgraph.checkpoint.memory import MemorySavermemory = MemorySaver()app = g.compile(checkpointer=memory)# 用 thread_id 区分不同会话config = {"configurable": {"thread_id": "1"}}app.invoke({"messages": [HumanMessage("我叫小明")]}, config)r = app.invoke({"messages": [HumanMessage("我叫什么名字？")]}, config)print(r["messages"][-1].content)   # 记得"小明"

Assistants API 核心概念：  Assistant  一个"助手"配置（指令+模型+工具）  Thread     一段会话的"线程"（装消息）  Message    一条消息（用户/助手）  Run        让助手跑一次（可异步）

          GPTs         Coze        Dify定位      闭源网页版    字节零代码    开源自部署知识库                          工作流    简单          强          强私有化               (云)       (自部署)适合      快速原型      国内快速产品  要源码/私有化