多 Agent 实战：任务边界、上下文传递与防止死循环

python -m pip install "langchain>=1,<2" "langchain-openai>=1,<2"
export OPENAI_API_KEY="替换为自己的密钥"

import json
from langchain.agents import create_agent
from langchain.tools import tool


MODEL = "openai:gpt-4.1-mini"


@tool
def get_device_status(device_id: str) -> str:
    """读取一个设备的当前状态，不执行写操作。"""
    devices = {"AX-3": {"status": "offline", "offline_minutes": 12}}
    return json.dumps(devices.get(device_id, {"status": "not_found"}), ensure_ascii=False)


@tool
def search_inspection_rule(keyword: str) -> str:
    """查询巡检规则，不执行写操作。"""
    return "设备离线超过 5 分钟时，建议创建人工巡检工单；创建前必须审批。"


device_expert = create_agent(
    MODEL,
    tools=[get_device_status],
    system_prompt="你是设备专家，只报告工具返回的事实，不猜测规则。",
)
rule_expert = create_agent(
    MODEL,
    tools=[search_inspection_rule],
    system_prompt="你是规则专家，只解释规则，不声称已经执行处置。",
)


def final_text(result: dict) -> str:
    content = result["messages"][-1].content
    return content if isinstance(content, str) else json.dumps(content, ensure_ascii=False)


@tool
def ask_device_expert(question: str) -> str:
    """向设备专家询问设备事实。每个任务最多调用一次。"""
    result = device_expert.invoke({"messages": [{"role": "user", "content": question}]})
    return final_text(result)


@tool
def ask_rule_expert(question: str) -> str:
    """向规则专家询问处置规则。每个任务最多调用一次。"""
    result = rule_expert.invoke({"messages": [{"role": "user", "content": question}]})
    return final_text(result)


supervisor = create_agent(
    MODEL,
    tools=[ask_device_expert, ask_rule_expert],
    system_prompt=(
        "你是巡检主管。先取得设备事实，再在需要时查询规则，最后汇总。"
        "同一专家最多调用一次，不得创建工单；需要写操作时仅返回待审批建议。"
    ),
)

result = supervisor.invoke(
    {"messages": [{"role": "user", "content": "检查 AX-3，并给出处置建议"}]},
    config={"recursion_limit": 8},
)
print(final_text(result))