用 Python 从零搭一个多智能体系统：从 Mock 闭环到真实 LLM 接入

用户需求 → ResearchAgent → ActivityAgent → BudgetAgent → FinalAgent → 最终行程

mkdir python-multi-agent-demo
cd python-multi-agent-demo
mkdir src

python-multi-agent-demo/
├── src/
│   ├── mock_agents.py
│   ├── workflow.py
│   └── openai_backend.py
└── .env.example

from dataclasses import dataclass


@dataclass
class AgentResult:
    agent_name: str
    output: str


@dataclass
class MockAgent:
    name: str
    role: str

    def run(self, task: str) -> AgentResult:
        if self.name == "Research Agent":
            output = (
                "目的地研究：伊斯坦布尔适合亲子游。推荐圣索菲亚大教堂、"
                "托普卡帕宫、伊斯坦布尔水族馆、KidZania 和博斯普鲁斯海峡游船。"
            )
        elif self.name == "Activity Planner Agent":
            output = (
                "活动安排：第 1 天游览历史城区；第 2 天安排水族馆和 KidZania；"
                "第 3 天安排公园、博物馆和集市。"
            )
        elif self.name == "Budget Agent":
            output = (
                "预算估算：4 人 3 天游约 3100-3800 美元，包含机票、住宿、餐饮、"
                "交通、门票和杂费。"
            )
        else:
            output = f"最终整合：\n{task}\n请根据以上信息生成一份简洁行程。"

        return AgentResult(agent_name=self.name, output=output)

from dataclasses import dataclass

from mock_agents import MockAgent


@dataclass
class TravelRequest:
    origin: str
    destination: str
    days: int
    travelers: int
    budget: str
    interests: str

    def to_prompt(self) -> str:
        return (
            f"出发地：{self.origin}\n"
            f"目的地：{self.destination}\n"
            f"天数：{self.days}\n"
            f"人数：{self.travelers}\n"
            f"预算：{self.budget}\n"
            f"偏好：{self.interests}"
        )


def build_agents() -> list[MockAgent]:
    return [
        MockAgent("Research Agent", "研究目的地、景点和旅行提示"),
        MockAgent("Activity Planner Agent", "根据研究结果安排每日活动"),
        MockAgent("Budget Agent", "估算旅行预算并给出省钱建议"),
        MockAgent("Final Travel Assistant", "整合所有结果，输出最终行程"),
    ]


def run_workflow(request: TravelRequest) -> str:
    current_context = request.to_prompt()
    history: list[str] = []

    for agent in build_agents():
        result = agent.run(current_context)
        history.append(f"## {result.agent_name}\n{result.output}")
        current_context = "\n\n".join(history)

    return current_context


if __name__ == "__main__":
    request = TravelRequest(
        origin="Islamabad",
        destination="Istanbul",
        days=3,
        travelers=4,
        budget="$4000",
        interests="kid friendly",
    )
    print(run_workflow(request))

python3 src/workflow.py

## Research Agent
目的地研究：...

## Activity Planner Agent
活动安排：...

## Budget Agent
预算估算：...

## Final Travel Assistant
最终整合：...

from dataclasses import dataclass
from typing import Protocol


class AgentBackend(Protocol):
    def generate(self, system_prompt: str, user_prompt: str) -> str:
        pass


@dataclass
class Agent:
    name: str
    role: str
    backend: AgentBackend

    def run(self, task: str) -> str:
        return self.backend.generate(
            system_prompt=self.role,
            user_prompt=task,
        )

OPENAI_BASE_URL=https://openrouter.ai/api/v1
OPENAI_API_KEY=replace-with-your-key
OPENAI_MODEL=gpt-4.1-mini

import os
from dataclasses import dataclass

from openai import OpenAI


@dataclass
class OpenAIBackend:
    base_url: str
    api_key: str
    model: str
    max_tokens: int = 1200

    @classmethod
    def from_env(cls) -> "OpenAIBackend":
        base_url = os.environ.get("OPENAI_BASE_URL", "https://openrouter.ai/api/v1")
        api_key = os.environ["OPENAI_API_KEY"]
        model = os.environ.get("OPENAI_MODEL", "gpt-4.1-mini")
        return cls(base_url=base_url, api_key=api_key, model=model)

    def generate(self, system_prompt: str, user_prompt: str) -> str:
        client = OpenAI(base_url=self.base_url, api_key=self.api_key)
        response = client.chat.completions.create(
            model=self.model,
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": user_prompt},
            ],
            max_tokens=self.max_tokens,
        )
        content = response.choices[0].message.content
        if not content:
            return ""
        return content

python3 -m venv .venv
. .venv/bin/activate
pip install openai

export OPENAI_BASE_URL="https://openrouter.ai/api/v1"
export OPENAI_API_KEY="你的 API Key"
export OPENAI_MODEL="gpt-4.1-mini"

from dataclasses import dataclass

from openai_backend import OpenAIBackend


@dataclass
class Agent:
    name: str
    role: str
    backend: OpenAIBackend

    def run(self, task: str) -> str:
        print(f"{self.name} is working...")
        return self.backend.generate(self.role, task)


def build_agents(backend: OpenAIBackend) -> list[Agent]:
    return [
        Agent(
            "Research Agent",
            "You are an expert travel researcher. Find attractions, hidden gems, local experiences, and travel tips.",
            backend,
        ),
        Agent(
            "Activity Planner Agent",
            "You are a travel activity planner. Create a day-wise plan from the research notes.",
            backend,
        ),
        Agent(
            "Budget Agent",
            "You estimate travel costs. Be concise and clearly mark estimates as estimates.",
            backend,
        ),
        Agent(
            "Final Travel Assistant",
            "You combine research, activities, and budget into a concise final itinerary.",
            backend,
        ),
    ]

{
  "attractions": ["Hagia Sophia", "Topkapi Palace"],
  "local_experiences": ["Bosphorus cruise"],
  "family_friendly_notes": ["KidZania Istanbul is suitable for children"],
  "risks": ["Opening hours and prices need real-time verification"]
}

{
  "currency": "USD",
  "estimated_total_min": 3100,
  "estimated_total_max": 3800,
  "assumptions": ["Flights are estimated, not real-time quotes"],
  "needs_verification": ["visa fees", "hotel prices", "flight prices"]
}

Final Travel Assistant → Validator Agent → Final Answer

You are a strict travel-plan validator.
Check the itinerary for unsupported claims, outdated travel data, budget inconsistencies, and overpacked schedules.
Do not rewrite the itinerary.
Return:
1. blocking_issues
2. warnings
3. facts_that_need_real_time_verification
4. recommended_user_disclaimer

{
  "ok": true,
  "agent": "Budget Agent",
  "output": "...",
  "warnings": [],
  "error": null
}

{
  "ok": false,
  "agent": "Budget Agent",
  "output": null,
  "warnings": [],
  "error": "model timeout"
}

只做预算，不改行程。
只做校验，不重写正文。
只做目的地研究，不生成最终计划。