第 8 章 给图装上手脚：工具调用

import os
from typing import Annotated, TypedDict
from dotenv import load_dotenv
from langchain_core.messages import AnyMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode

load_dotenv()

# ---- 1. 定义工具 ----
COMPANIES = {
    "深澜科技": {"scale": "500-1000人", "field": "跨境支付基础设施", "stack": "Java/Go 为主"},
    "星河互联": {"scale": "2000+人", "field": "本地生活服务平台", "stack": "Python 微服务"},
}
JOBS = {
    "python-backend": {
        "title": "Python 后端工程师",
        "requirements": "5 年以上经验；精通 Python/FastAPI；有高并发系统经验",
    },
}

@tool
def lookup_company(name: str) -> str:
    """查询指定公司的背景信息：规模、业务领域、技术栈。

    Args:
        name: 公司全名，例如「深澜科技」
    """
    info = COMPANIES.get(name)
    return f"{name}：{info}" if info else f"未查询到公司「{name}」的背景信息"

@tool
def get_job_requirements(job_id: str) -> str:
    """按岗位 ID 查询最新的岗位 JD。

    Args:
        job_id: 岗位标识，例如「python-backend」
    """
    job = JOBS.get(job_id)
    return f"{job['title']} 的要求：{job['requirements']}" if job else f"岗位 {job_id} 不存在"

tools = [lookup_company, get_job_requirements]

# ---- 2. 状态：消息列表（工具调用是消息层面的协作） ----
class AgentState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]

# ---- 3. model 节点 ----
llm = ChatOpenAI(
    model=os.getenv("LLM_MODEL", "gpt-4o-mini"),
    temperature=0,
).bind_tools(tools)                       # 关键：把工具说明绑定给模型

SYSTEM_PROMPT = """你是资深技术招聘官，负责评估候选人简历。
评估前，如需了解候选人曾任职公司的背景或岗位最新要求，主动调用相应工具。
信息充分后，输出最终评估结论（匹配度 + 三条理由）。"""

def call_model(state: AgentState) -> dict:
    from langchain_core.messages import SystemMessage
    messages = [SystemMessage(SYSTEM_PROMPT)] + state["messages"]
    response = llm.invoke(messages)
    return {"messages": [response]}       # add_messages：追加

# ---- 4. 路由函数：模型是否要调工具 ----
def route_after_model(state: AgentState) -> str:
    last = state["messages"][-1]          # 最新一条是模型的回复
    return "tools" if last.tool_calls else END

# ---- 5. 组装图 ----
builder = StateGraph(AgentState)
builder.add_node("model", call_model)
builder.add_node("tools", ToolNode(tools))   # 内置节点：执行所有 tool_calls
builder.add_edge(START, "model")
builder.add_conditional_edges("model", route_after_model, {"tools": "tools", END: END})
builder.add_edge("tools", "model")           # 回边：循环闭合
graph = builder.compile()

# ---- 6. 运行 ----
from langchain_core.messages import HumanMessage

result = graph.invoke({
    "messages": [HumanMessage(
        "候选人李思远：6 年后端经验，Python/FastAPI/PostgreSQL，"
        "此前在深澜科技负责支付网关。请对照 python-backend 岗位评估。"
    )]
})
print(result["messages"][-1].content)

[HumanMessage]                                用户请求
[AIMessage(tool_calls=[lookup_company])]      模型：先查公司
[ToolMessage(公司背景)]                        ToolNode 执行结果
[AIMessage(tool_calls=[get_job_requirements])] 模型：再查 JD
[ToolMessage(JD 内容)]                         ToolNode 执行结果
[AIMessage(最终评估结论)]                      模型：信息够了，收工

from typing import Annotated
from typing_extensions import NotRequired
from langchain_core.tools import InjectedToolArg
from langgraph.prebuilt import InjectedState

class AgentState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]
    current_user: NotRequired[str]        # 不想让模型染指的字段

@tool
def submit_evaluation(
    conclusion: str,                       # 模型提供
    state: Annotated[dict, InjectedState], # 引擎注入：模型看不见此参数
) -> str:
    """提交最终评估结论。

    Args:
        conclusion: 评估结论文本
    """
    user = state["current_user"]
    return f"已代表 {user} 提交评估：{conclusion}"