┌────────────────────────────────────────────────┐
│  Deep Agents —— "开箱即用的完整 Agent"          │  一行 create_deep_agent 起步
├────────────────────────────────────────────────┤
│  LangGraph —— "画流程图的运行时"                 │  节点、边、状态、子图、Command
├────────────────────────────────────────────────┤
│  LangChain —— "基础积木盒"                       │  Models / Tools / Prompts / Messages
└────────────────────────────────────────────────┘

RESEARCH_AGENT_NAME = "ResearchAgent"

RESEARCH_AGENT_SYSTEM_PROMPT = """
你是一名 ResearchAgent。你的职责：
- 接收一个研究问题
- 调用合适的工具收集资料
- 最终输出结构化报告

你不做的事：
- 不执行代码
- 不直接修改文件
- 不回答与研究无关的问题（如闲聊）
"""

# 在 LangGraph 中体现为：State 里有一个字段标记当前 agent 名字，
# 每个节点的 system_prompt 都引用 RESEARCH_AGENT_SYSTEM_PROMPT。
```text
# 三、第二步：定义 Agent 需要的工具、技能

## 3.1 什么是"工具"

工具是 Agent 能调用的**普通 Python 函数**。LangGraph 通过 `@tool` 装饰器把函数包装成 LLM 能"看见"的工具（带 schema）。

## 3.2 工具的 3 个要素

1. **函数名**：LLM 用它来调用（snake_case）
2. **文档字符串**：告诉 LLM 这个工具能干嘛、何时该用
3. **类型注解**：参数和返回值的类型（自动生成 schema）

## 3.3 具体例子：3 个常用工具

### 例子 1：天气查询工具

```python
from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """查询指定城市的当前天气。

    Args:
        city: 城市名，如 "北京" / "上海" / "Tokyo"。
    """
    # 真实场景：调天气 API。这里用伪数据演示。
    return json.dumps({"city": city, "temperature": 24, "condition": "晴"})

@tool
def calculate(expression: str) -> str:
    """对数学表达式求值。

    支持 + - * / () 和数字。不支持变量、函数、负数。
    Args:
        expression: 如 "(25+17)*3"。
    """
    # 安全：白名单只允许数字和运算符
    allowed = set("0123456789+-*/().% ")
    if not all(ch in allowed for ch in expression):
        return "错误：表达式包含非法字符"
    try:
        return str(eval(expression, {"__builtins__": {}}, {}))
    except Exception as e:
        return f"计算失败: {e}"

import datetime as _dt

@tool
def get_current_time() -> str:
    """获取当前的服务器时间，返回格式 'YYYY-MM-DD HH:MM:SS'。"""
    return _dt.datetime.now().strftime("%Y-%m-%d %H:%M:%S")

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="MiniMax-M3", ...)

# 1) 定义工具
TOOLS = [get_weather, calculate, get_current_time]

# 2) 把工具告诉 LLM
llm_with_tools = llm.bind_tools(TOOLS)

# 3) LLM 看到用户问题时，会自动选工具
response = llm_with_tools.invoke("北京天气怎么样？")
print(response.tool_calls)
# 输出：[{'name': 'get_weather', 'args': {'city': '北京'}, 'id': '...'}]

@tool
def do_everything(action: str, data1: str = "", data2: str = "", data3: str = "") -> str:
    """万能工具，能做各种事"""
    if action == "weather":
        ...
    elif action == "calc":
        ...
    elif action == "time":
        ...
    # 50 行 if-elif ...

        ┌─────────────┐
        │   START     │
        └──────┬──────┘
               ▼
        ┌─────────────┐
        │  llm_call   │  调用 LLM，让它决定要不要调工具
        └──────┬──────┘
               │
       ┌───────┴───────┐
  有 tool_calls?      没 tool_calls?
       │                    │
       ▼                    ▼
  ┌─────────┐         ┌─────────┐
  │tool_node│         │   END   │
  │执行工具 │         │ 结束流程│
  └────┬────┘         └─────────┘
       │
       └─ 把结果回灌给 llm_call ─┐
                                 │
                            (回到 llm_call)

[HumanMessage]    北京天气怎么样？顺便帮我算一下 (25+17)*3 等于多少？
[AIMessage]      tool_calls=[
                   get_weather(city="北京"),
                   calculate(expression="(25+17)*3")
                 ]
[ToolMessage]    {"city": "北京", "temperature": 21, "condition": "晴"}
[ToolMessage]    126
[AIMessage]      北京现在 21 度，天气晴朗；(25+17)*3 = 126。还有别的需要吗？

                        用户
                         │
                         ▼
                ┌────────────────┐
                │  Supervisor    │
                │  (决定派给谁)  │
                └────────┬───────┘
                         │
       ┌─────────┬───────┴───────┬─────────┐
       ▼         ▼               ▼         ▼
  researcher  writer          reviewer   END
  (查资料)    (写文章)        (评审)    (结束)
       │         │               │
       └─────────┴───────┬───────┘
                         │
                  (回灌给 Supervisor)

from langgraph.types import Command

# 把"派活"建模成 4 个工具
@tool
def transfer_to_researcher(task: str) -> str:
    """派给 Researcher"""
    return f"派给 Researcher：{task}"

@tool
def transfer_to_writer(task: str) -> str:
    """派给 Writer"""
    return f"派给 Writer：{task}"

@tool
def transfer_to_reviewer(task: str) -> str:
    """派给 Reviewer"""
    return f"派给 Reviewer：{task}"

@tool
def finish(answer: str) -> str:
    """任务完成，结束流程"""
    return f"FINAL::{answer}"

supervisor_llm = base_llm.bind_tools([
    transfer_to_researcher, transfer_to_writer,
    transfer_to_reviewer,   finish,
])

def supervisor_node(state) -> Command:
    """Supervisor：看一眼白板，决定派给谁"""
    ai = supervisor_llm.invoke([
        SystemMessage(content="你是 Supervisor..."),
        *state["messages"]
    ])

    # 把 LLM 选的"工具名"翻译成"跳转指令"
    goto = END
    if ai.tool_calls:
        name = ai.tool_calls[0]["name"]
        goto = {
            "transfer_to_researcher": "researcher",
            "transfer_to_writer":     "writer",
            "transfer_to_reviewer":   "reviewer",
            "finish":                 END,
        }[name]

    return Command(goto=goto, update={"messages": [ai]})

def make_worker_node(name, system_prompt, tools, tool_node):
    """工厂函数：给每个 Worker 生成一个节点"""
    worker_llm = base_llm.bind_tools(tools)

    # Worker 内部子图：agent ↔ tools 循环
    subgraph = (
        StateGraph(MessagesState)
        .add_node("agent", lambda s: {"messages": [
            worker_llm.invoke([SystemMessage(system_prompt), *s["messages"]])
        ]})
        .add_node("tools", tool_node)
        .add_edge(START, "agent")
        .add_conditional_edges("agent",
            lambda s: "tools" if s["messages"][-1].tool_calls else END)
        .add_edge("tools", "agent")
        .compile()
    )

    def worker_node(state: MessagesState) -> Command[Literal["supervisor"]]:
        # 1) 从 Supervisor 的派活指令里取出 task
        supervisor_msg = next(m for m in reversed(state["messages"])
                              if isinstance(m, AIMessage))
        task_text = supervisor_msg.tool_calls[0]["args"].get("task", "")
        tool_call_id = supervisor_msg.tool_calls[0]["id"]

        # 2) 跑 Worker 子图
        result = subgraph.invoke({
            "messages": [HumanMessage(content=task_text)]
        })

        # 3) ToolMessage 紧跟在 Supervisor AIMessage 之后
        transfer_tm = ToolMessage(
            content=f"{name} 完成：{result['messages'][-1].content}",
            name=f"transfer_to_{name}",
            tool_call_id=tool_call_id,
        )
        return Command(
            goto="supervisor",
            update={"messages": [transfer_tm, *result["messages"]]},
        )

    return worker_node

# 生成 3 个 Worker
researcher_node = make_worker_node("researcher", "你是研究员...", RESEARCH_TOOLS, RESEARCH_TOOL_NODE)
writer_node     = make_worker_node("writer",     "你是写手...",   WRITER_TOOLS,     WRITER_TOOL_NODE)
reviewer_node   = make_worker_node("reviewer",   "你是评审员...", REVIEWER_TOOLS,   REVIEWER_TOOL_NODE)

# 装配主图
builder = StateGraph(MessagesState)
builder.add_node("supervisor", supervisor_node)
builder.add_node("researcher", researcher_node)
builder.add_node("writer",     writer_node)
builder.add_node("reviewer",   reviewer_node)
builder.add_edge(START, "supervisor")
graph = builder.compile()

[Human]   请帮我写一篇 300 字左右的短文，介绍 LangGraph 在企业级 AI Agent 中的应用价值。

[Supervisor]   transfer_to_researcher(task="收集 LangGraph 资料...")
[Researcher]   web_search("LangGraph 是什么..."), web_search("LangGraph features...")...
[Researcher]   整理出关键要点...

[Supervisor]   transfer_to_writer(task="基于资料写 280~350 字短文")
[Writer]       # LangGraph：企业级 AI Agent 的图驱动编排框架 ...（约 624 字）

[Supervisor]   transfer_to_reviewer(task="检查字数和内容")
[Reviewer]     字数 624，超标！建议精简到 280~350。

[Supervisor]   transfer_to_writer(task="精简到 280~350 字")
[Writer]       ...（精简版）
[Reviewer]     字数 375，再精简。

[Supervisor]   transfer_to_writer(task="再精简到 350 以内")
[Writer]       ...（最终版）
[Reviewer]     PASS！350 字以内，结构清晰。

[Supervisor]   finish(answer="## 任务完成！...")

# 编译后直接画图
agent = builder.compile()
from IPython.display import Image, display
display(Image(agent.get_graph().draw_mermaid_png()))

# stream 模式：每执行完一个节点就返回一个 snapshot
for step in agent.stream({"messages": [HumanMessage(content="...")]}, stream_mode="values"):
    step["messages"][-1].pretty_print()

pip install langsmith
export LANGSMITH_TRACING=true
export LANGSMITH_API_KEY=lsv2_xxx

openai.BadRequestError: 400 - An assistant message with 'tool_calls' must be followed
by tool messages responding to each 'tool_call_id'.

# 错误写法：先放子图消息，再放 ToolMessage
update={"messages": [*result["messages"], ToolMessage(...)]}

# 正确写法：ToolMessage 必须紧跟在 Supervisor AIMessage 后
update={"messages": [ToolMessage(...), *result["messages"]]}

openai.BadRequestError: 400 - An assistant message with 'tool_calls' must be followed
by tool messages responding to each 'tool_call_id'.

# Worker 启动时只构造一个新 HumanMessage，丢弃 Supervisor 的派活 AIMessage
def worker_node(state):
    supervisor_msg = next(m for m in reversed(state["messages"])
                          if isinstance(m, AIMessage))
    task_text = supervisor_msg.tool_calls[0]["args"].get("task", "")

    # 用全新 HumanMessage 启动子图，不传 state["messages"]
    result = subgraph.invoke({"messages": [HumanMessage(content=task_text)]})
    ...

from deepagents import create_deep_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
import json, os

# 1) 你的工具
@tool
def web_search(query: str) -> str:
    """联网搜索"""
    return json.dumps({"query": query, "results": [...]}, ensure_ascii=False)

@tool
def count_words(text: str) -> str:
    """统计字数"""
    return str(len(text))

# 2) LLM
llm = ChatOpenAI(model="MiniMax-M3", base_url=..., api_key=...)

# 3) 一行创建
agent = create_deep_agent(
    model=llm,
    tools=[web_search, count_words],
    system_prompt=(
        "你是研究员 + 写手。\n"
        "工作流：\n"
        "1) write_todos 拆任务\n"
        "2) web_search 查资料\n"
        "3) 写文章，用 count_words 自检字数\n"
    ),
)

# 4) 运行
result = agent.invoke({"messages": [{"role": "user", "content": "..."}]})
print(result["messages"][-1].content)

pip install langgraph langchain-openai langchain-core deepagents python-dotenv

MINIMAX_API_KEY=sk-xxx
MINIMAX_BASE_URL=https://api.minimaxi.com/v1
MINIMAX_MODEL=MiniMax-M3