5W字长文 Agent多智能体探秘：架构设计、交互模式与应用实践深度剖析

❝

本文 5W字符 将剖析单代理与多代理架构的差异，详解多代理系统中的多种模式，如并行、顺序等，还会探讨代理间的通信机制。
更过agent系列文章：
1. AI 智能体，颠覆还是赋能？一文读懂！
2. 1W8000字 解锁 AI 高效运作密码：工作流与智能体如何协同？
3. 万字深度剖析 AI 代理：类型、应用及优势与局限全解析
4. 1W5000字 深度剖析大模型Agent框架
5. Agent系列 1W字用Python从零搭建AI智能体
6. 万字深度剖析 AI 代理：类型、应用及优势与局限全解析

一、引言

在人工智能领域，代理是一类借助大语言模型（LLM）来决定应用程序控制流的系统。随着开发的推进，这类系统往往会变得愈发复杂，给管理和扩展带来诸多难题。比如，你可能会遭遇以下状况：

• 工具选择困境：代理可调用的工具繁多，导致在决策下一步使用哪个工具时表现欠佳。
• 上下文管理难题：上下文信息过于繁杂，单个代理难以有效追踪和处理。
• 专业领域需求多样：系统内需要涵盖多个专业领域，像规划师、研究员、数学专家等角色。

为化解这些问题，一种有效的策略是把应用程序拆分成多个较小的独立代理，进而组合成多代理系统。这些独立代理的复杂程度差异较大，简单的可能仅涉及一个提示和一次LLM调用，复杂的则可能像ReAct代理那般（甚至更复杂！）。

随着代理框架的蓬勃发展，众多公司纷纷构建自己的多代理系统，期望找到解决所有代理任务的通用方案。两年前，研究人员设计出一款名为ChatDev的多代理协作系统。ChatDev宛如一家虚拟软件公司，通过各类智能代理来运作，这些代理分别承担着首席执行官、首席产品官、美工、编码器、审阅者、测试人员等不同角色，与常规的软件工程公司极为相似。

所有这些代理协同作业、相互通信，成功开发出一款视频游戏。这一成果让不少人认为，任何软件工程任务都能借助这种多代理架构完成，其中每个AI都有着独特的分工。然而，实际的实验表明，并非所有问题都能靠同一架构解决。在某些情形下，更为简单的架构反而能提供更高效、成本效益更好的解决方案。

1.1 单代理与多代理架构

起初，单代理方法看似可行（即一个人工智能代理能处理从浏览器导航到文件操作等所有事务）。但随着时间推移，任务复杂度攀升，工具数量增多，单代理模式便开始面临挑战。

当代理出现异常行为时，我们会察觉到一些影响，这可能源于以下因素：

• 工具过载：工具数量过多，使得代理在选择工具及其使用时机时陷入困惑。
• 上下文臃肿：代理的上下文窗口不断扩大，其中包含的工具数量过多。
• 错误频发：由于职责范围过宽，代理开始产生次优甚至错误的结果。

当我们着手自动化多个不同的子任务（如数据提取、报告生成）时，或许就该考虑划分职责了。通过采用多个AI代理，每个代理专注于自身的领域和工具包，这样不仅能提升解决方案的清晰度和质量，还能简化代理的开发流程。

二、多智能体架构

正如您所看到的，单代理和多代理架构既有优势也有劣势。当任务简单且定义明确，并且没有特定资源限制时，单代理架构是理想之选。另一方面，当用例复杂且动态，需要更专业的知识和协作，或者具有可扩展性和适应性要求时，多代理架构则能发挥很大的作用。

2.1 多智能体系统中的模式

在多代理系统中，连接代理的方式有好几种：

2.1.1 并行

多个代理同时处理任务的不同部分。

示例：我们希望使用3个代理同时对给定文本进行总结、翻译和情感分析。

代码：

from typing import Dict, Any, TypedDict
from langgraph.graph import StateGraph, END
from langchain_core.runnables import RunnableConfig
from textblob import TextBlob
import re
import time


class AgentState(TypedDict):
    text: str
    summary: str
    translation: str
    sentiment: str
    summary_time: float
    translation_time: float
    sentiment_time: float


def summarize_agent(state: AgentState) -> Dict[str, Any]:
    print("Summarization Agent: Running")
    start_time = time.time()
    try:
        text = state["text"]
        ifnot text.strip():
            return {
                "summary": "No text provided for summarization.",
                "summary_time": 0.0
            }
        time.sleep(2)
        sentences = re.split(r'(?<=[.!?]) +', text.strip())
        scored_sentences = [(s, len(s.split())) for s in sentences if s]
        top_sentences = [s for s, _ in sorted(scored_sentences, key=lambda x: x[1], reverse=True)[:2]]
        summary = " ".join(top_sentences) if top_sentences else"Text too short to summarize."
        processing_time = time.time() - start_time
        print(f"Summarization Agent: Completed in {processing_time:.2f} seconds")
        return {
            "summary": summary,
            "summary_time": processing_time
        }
    except Exception as e:
        return {
            "summary": f"Error in summarization: {str(e)}",
            "summary_time": 0.0
        }


def translate_agent(state: AgentState) -> Dict[str, Any]:
    print("Translation Agent: Running")
    start_time = time.time()
    try:
        text = state["text"]
        ifnot text.strip():
            return {
                "translation": "No text provided for translation.",
                "translation_time": 0.0
            }
        time.sleep(3)
        translation = (
            "El nuevo parque en la ciudad es una maravillosa adición. "
            "Las familias disfrutan de los espacios abiertos, y a los niños les encanta el parque infantil. "
            "Sin embargo, algunas personas piensan que el área de estacionamiento es demasiado pequeña."
        )
        processing_time = time.time() - start_time
        print(f"Translation Agent: Completed in {processing_time:.2f} seconds")
        return {
            "translation": translation,
            "translation_time": processing_time
        }
    except Exception as e:
        return {
            "translation": f"Error in translation: {str(e)}",
            "translation_time": 0.0
        }


def sentiment_agent(state: AgentState) -> Dict[str, Any]:
    print("Sentiment Agent: Running")
    start_time = time.time()
    try:
        text = state["text"]
        ifnot text.strip():
            return {
                "sentiment": "No text provided for sentiment analysis.",
                "sentiment_time": 0.0
            }
        time.sleep(1.5)
        blob = TextBlob(text)
        polarity = blob.sentiment.polarity
        subjectivity = blob.sentiment.subjectivity
        sentiment = "Positive"if polarity > 0else"Negative"if polarity < 0else"Neutral"
        result = f"{sentiment} (Polarity: {polarity:.2f}, Subjectivity: {subjectivity:.2f})"
        processing_time = time.time() - start_time
        print(f"Sentiment Agent: Completed in {processing_time:.2f} seconds")
        return {
            "sentiment": result,
            "sentiment_time": processing_time
        }
    except Exception as e:
        return {
            "sentiment": f"Error in sentiment analysis: {str(e)}",
            "sentiment_time": 0.0
        }


def join_parallel_results(state: AgentState) -> AgentState:
    return state


def build_parallel_graph() -> StateGraph:
    workflow = StateGraph(AgentState)
    parallel_branches = {
        "summarize_node": summarize_agent,
        "translate_node": translate_agent,
        "sentiment_node": sentiment_agent
    }
    for name, agent in parallel_branches.items():
        workflow.add_node(name, agent)
    workflow.add_node("branch", lambda state: state)
    workflow.add_node("join", join_parallel_results)
    workflow.set_entry_point("branch")
    for name in parallel_branches:
        workflow.add_edge("branch", name)
        workflow.add_edge(name, "join")
    workflow.add_edge("join", END)
    return workflow.compile()


def main():
    text = (
        "The new park in the city is a wonderful addition. Families are enjoying the open spaces, "
        "and children love the playground. However, some people think the parking area is too small."
    )
    initial_state: AgentState = {
        "text": text,
        "summary": "",
        "translation": "",
        "sentiment": "",
        "summary_time": 0.0,
        "translation_time": 0.0,
        "sentiment_time": 0.0
    }
    print("nBuilding new graph...")
    app = build_parallel_graph()
    print("nStarting parallel processing...")
    start_time = time.time()
    config = RunnableConfig(parallel=True)
    result = app.invoke(initial_state, config=config)
    total_time = time.time() - start_time
    print("n=== Parallel Task Results ===")
    print(f"Input Text:n{text}n")
    print(f"Summary:n{result['summary']}n")
    print(f"Translation (Spanish):n{result['translation']}n")
    print(f"Sentiment Analysis:n{result['sentiment']}n")
    print("n=== Processing Times ===")
    processing_times = {
        "summary": result["summary_time"],
        "translation": result["translation_time"],
        "sentiment": result["sentiment_time"]
    }
    for agent, time_taken in processing_times.items():
        print(f"{agent.capitalize()}: {time_taken:.2f} seconds")
    print(f"nTotal Wall Clock Time: {total_time:.2f} seconds")
    print(f"Sum of Individual Processing Times: {sum(processing_times.values()):.2f} seconds")
    print(f"Time Saved by Parallel Processing: {sum(processing_times.values()) - total_time:.2f} seconds")


if __name__ == "__main__":
    main()

输出：

Building new graph...
Starting parallel processing...
Sentiment Agent: Running
Summarization Agent: Running
Translation Agent: Running
Sentiment Agent: Completed in 1.50 seconds
Summarization Agent: Completed in 2.00 seconds
Translation Agent: Completed in 3.00 seconds
=== Parallel Task Results ===
Input Text:
The new park in the city is a wonderful addition. Families are enjoying the open spaces, and children love the playground. However, some people think the parking area is too small.
Summary:
Families are enjoying the open spaces, and children love the playground. The new park in the city is a wonderful addition.
Translation (Spanish):
El nuevo parque en la ciudad es una maravillosa adición. Las familias disfrutan de los espacios abiertos, y a los niños les encanta el parque infantil. Sin embargo, algunas personas piensan que el área de estacionamiento es demasiado pequeña.
Sentiment Analysis:
Positive (Polarity: 0.31, Subjectivity: 0.59)
=== Processing Times ===
Summary: 2.00 seconds
Translation: 3.00 seconds
Sentiment: 1.50 seconds
Total Wall Clock Time: 3.01 seconds
Sum of Individual Processing Times: 6.50 seconds
Time Saved by Parallel Processing: 3.50 seconds

并行模式特点：

• 并行性：三个任务（总结、翻译、情感分析）同时运行，减少了总处理时间。
• 独立性：每个代理独立处理输入文本，执行过程中无需代理间通信。
• 协调性：队列确保结果被安全收集并按顺序显示。
• 实际用例：总结、翻译和情感分析是常见的自然语言处理（NLP）任务，并行处理对较大文本特别有益。

2.1.2 顺序

任务按顺序依次处理，前一个代理的输出成为下一个代理的输入。

示例：多步审批流程。

代码：

from typing import Dict
from langgraph.graph import StateGraph, MessagesState, END
from langchain_core.runnables import RunnableConfig
from langchain_core.messages import HumanMessage, AIMessage
import json


def team_lead_agent(state: MessagesState, config: RunnableConfig) -> Dict:
    print("Agent (Team Lead): Starting review")
    messages = state["messages"]
    proposal = json.loads(messages[0].content)
    title = proposal.get("title", "")
    amount = proposal.get("amount", 0.0)
    ifnot title or amount <= 0:
        status = "Rejected"
        comment = "Team Lead: Proposal rejected due to missing title or invalid amount."
        goto = END
    else:
        status = "Approved by Team Lead"
        comment = "Team Lead: Proposal is complete and approved."
        goto = "dept_manager"
    print(f"Agent (Team Lead): Review complete - {status}")
    messages.append(AIMessage(
        content=json.dumps({"status": status, "comment": comment}),
        additional_kwargs={"agent": "team_lead", "goto": goto}
    ))
    return {"messages": messages}


def dept_manager_agent(state: MessagesState, config: RunnableConfig) -> Dict:
    print("Agent (Department Manager): Starting review")
    messages = state["messages"]
    team_lead_msg = next((m for m in messages if m.additional_kwargs.get("agent") == "team_lead"), None)
    proposal = json.loads(messages[0].content)
    amount = proposal.get("amount", 0.0)
    if json.loads(team_lead_msg.content)["status"] != "Approved by Team Lead":
        status = "Rejected"
        comment = "Department Manager: Skipped due to Team Lead rejection."
        goto = END
    elif amount > 100000:
        status = "Rejected"
        comment = "Department Manager: Budget exceeds limit."
        goto = END
    else:
        status = "Approved by Department Manager"
        comment = "Department Manager: Budget is within limits."
        goto = "finance_director"
    print(f"Agent (Department Manager): Review complete - {status}")
    messages.append(AIMessage(
        content=json.dumps({"status": status, "comment": comment}),
        additional_kwargs={"agent": "dept_manager", "goto": goto}
    ))
    return {"messages": messages}


def finance_director_agent(state: MessagesState, config: RunnableConfig) -> Dict:
    print("Agent (Finance Director): Starting review")
    messages = state["messages"]
    dept_msg = next((m for m in messages if m.additional_kwargs.get("agent") == "dept_manager"), None)
    proposal = json.loads(messages[0].content)
    amount = proposal.get("amount", 0.0)
    if json.loads(dept_msg.content)["status"] != "Approved by Department Manager":
        status = "Rejected"
        comment = "Finance Director: Skipped due to Dept Manager rejection."
    elif amount > 50000:
        status = "Rejected"
        comment = "Finance Director: Insufficient budget."
    else:
        status = "Approved"
        comment = "Finance Director: Approved and feasible."
    print(f"Agent (Finance Director): Review complete - {status}")
    messages.append(AIMessage(
        content=json.dumps({"status": status, "comment": comment}),
        additional_kwargs={"agent": "finance_director", "goto": END}
    ))
    return {"messages": messages}


def route_step(state: MessagesState) -> str:
    for msg in reversed(state["messages"]):
        goto = msg.additional_kwargs.get("goto")
        if goto:
            print(f"Routing: Agent {msg.additional_kwargs.get('agent')} set goto to {goto}")
            return goto
    return END


builder = StateGraph(MessagesState)
builder.add_node("team_lead", team_lead_agent)
builder.add_node("dept_manager", dept_manager_agent)
builder.add_node("finance_director", finance_director_agent)
builder.set_entry_point("team_lead")
builder.add_conditional_edges("team_lead", route_step, {
    "dept_manager": "dept_manager",
    END: END
})
builder.add_conditional_edges("dept_manager", route_step, {
    "finance_director": "finance_director",
    END: END
})
builder.add_conditional_edges("finance_director", route_step, {
    END: END
})
workflow = builder.compile()


def main():
    initial_state = {
        "messages": [
            HumanMessage(
                content=json.dumps({
                    "title": "New Equipment Purchase",
                    "amount": 40000.0,
                    "department": "Engineering"
                })
            )
        ]
    }
    result = workflow.invoke(initial_state)
    messages = result["messages"]
    proposal = json.loads(messages[0].content)
    print("n=== Approval Results ===")
    print(f"Proposal Title: {proposal['title']}")
    final_status = "Unknown"
    comments = []
    for msg in messages[1:]:
        if isinstance(msg, AIMessage):
            try:
                data = json.loads(msg.content)
                if"status"in data:
                    final_status = data["status"]
                if"comment"in data:
                    comments.append(data["comment"])
            except Exception:
                continue
    print(f"Final Status: {final_status}")
    print("Comments:")
    for comment in comments:
        print(f"  - {comment}")


if __name__ == "__main__":
    main()

输出（金额 = 40,000美元）：

Agent (Team Lead): Starting review
Agent (Team Lead): Review complete - Approved by Team Lead
Routing: Agent team_lead set goto to dept_manager
Agent (Department Manager): Starting review
Agent (Department Manager): Review complete - Approved by Department Manager
Routing: Agent dept_manager set goto to finance_director
Agent (Finance Director): Starting review
Agent (Finance Director): Review complete - Approved
Routing: Agent finance_director set goto to __end__
=== Approval Results ===
Proposal Title: New Equipment Purchase
Final Status: Approved
Comments:
- Team Lead: Proposal is complete and approved.
- Department Manager: Budget is within limits.
- Finance Director: Approved and feasible.

顺序执行特点：

• 顺序执行：代理按顺序运行：团队领导 → 部门经理 → 财务总监。
• 中断机制：如果任何一个代理拒绝，循环将中断，跳过剩余代理。
• 对象修改：每个代理修改共享的提案对象，更新状态和评论。

协调方式：

• 结果存储在列表中，但提案对象在代理之间传递状态。
• 不使用多进程，确保单线程、有序的工作流程。

2.1.3 循环

代理以迭代循环的方式运作，基于其他代理的反馈持续改进它们的输出。

示例：评估用例，如代码编写与代码测试。

代码：

from typing import Dict, Any, List
from langgraph.graph import StateGraph, END
from langchain_core.runnables import RunnableConfig
import textwrap


class EvaluationState(Dict[str, Any]):
    code: str = ""
    feedback: str = ""
    passed: bool = False
    iteration: int = 0
    max_iterations: int = 3
    history: List[Dict] = []

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.setdefault("code", "")
        self.setdefault("feedback", "")
        self.setdefault("passed", False)
        self.setdefault("iteration", 0)
        self.setdefault("max_iterations", 3)
        self.setdefault("history", [])


def code_writer_agent(state: EvaluationState, config: RunnableConfig) -> Dict[str, Any]:
    print(f"Iteration {state['iteration'] + 1} - Code Writer: Generating code")
    print(f"Iteration {state['iteration'] + 1} - Code Writer: Received feedback: {state['feedback']}")
    iteration = state["iteration"] + 1
    feedback = state["feedback"]
    if iteration == 1:
        code = textwrap.dedent("""
def factorial(n):
    result = 1
    for i in range(1, n + 1):
        result *= i
    return result
""")
        writer_feedback = "Initial code generated."
    elif"factorial(0)"in feedback.lower():
        code = textwrap.dedent("""
def factorial(n):
    if n == 0:
        return 1
    result = 1
    for i in range(1, n + 1):
        result *= i
    return result
""")
        writer_feedback = "Fixed handling for n=0."
    elif"factorial(-1)"in feedback.lower() or"negative"in feedback.lower():
        code = textwrap.dedent("""
def factorial(n):
    if n < 0:
        raise ValueError("Factorial not defined for negative numbers")
    if n == 0:
        return 1
    result = 1
    for i in range(1, n + 1):
        result *= i
    return result
""")
        writer_feedback = "Added error handling for negative inputs."
    else:
        code = state["code"]
        writer_feedback = "No further improvements identified."
    print(f"Iteration {iteration} - Code Writer: Code generated")
    return {
        "code": code,
        "feedback": writer_feedback,
        "iteration": iteration
    }


def code_tester_agent(state: EvaluationState, config: RunnableConfig) -> Dict[str, Any]:
    print(f"Iteration {state['iteration']} - Code Tester: Testing code")
    code = state["code"]
    try:
        test_cases = [
            (0, 1),
            (1, 1),
            (5, 120),
            (-1, None),
        ]
        namespace = {}
        exec(code, namespace)
        factorial = namespace.get('factorial')
        ifnot callable(factorial):
            return {"passed": False, "feedback": "No factorial function found."}
        feedback_parts = []
        passed = True
        for input_val, expected in test_cases:
            try:
                result = factorial(input_val)
                if expected isNone:
                    passed = False
                    feedback_parts.append(f"Test failed: factorial({input_val}) should raise an error.")
                elif result != expected:
                    passed = False
                    feedback_parts.append(f"Test failed: factorial({input_val}) returned {result}, expected {expected}.")
            except ValueError as ve:
                if expected isnotNone:
                    passed = False
                    feedback_parts.append(f"Test failed: factorial({input_val}) raised ValueError unexpectedly: {str(ve)}")
            except Exception as e:
                passed = False
                feedback_parts.append(f"Test failed: factorial({input_val}) caused error: {str(e)}")
        feedback = "All tests passed!"if passed else"n".join(feedback_parts)
        print(f"Iteration {state['iteration']} - Code Tester: Testing complete - {'Passed' if passed else 'Failed'}")
        history = state["history"]
        history.append({
            "iteration": state["iteration"],
            "code": code,
            "feedback": feedback,
            "passed": passed
        })
        return {
            "passed": passed,
            "feedback": feedback,
            "history": history
        }
    except Exception as e:
        print(f"Iteration {state['iteration']} - Code Tester: Failed")
        return {"passed": False, "feedback": f"Error in testing: {str(e)}"}


def should_continue(state: EvaluationState) -> str:
    if state["passed"] or state["iteration"] >= state["max_iterations"]:
        print(f"Iteration {state['iteration']} - {'Loop stops: Tests passed' if state['passed'] else 'Loop stops: Max iterations reached'}")
        return"end"
    print(f"Iteration {state['iteration']} - Loop continues: Tests failed")
    return"code_writer"


workflow = StateGraph(EvaluationState)
workflow.add_node("code_writer", code_writer_agent)
workflow.add_node("code_tester", code_tester_agent)
workflow.set_entry_point("code_writer")
workflow.add_edge("code_writer", "code_tester")
workflow.add_conditional_edges(
    "code_tester",
    should_continue,
    "code_writer": "code_writer",
    "end": END
)
app = workflow.compile()


def main():
    initial_state = EvaluationState()
    result = app.invoke(initial_state)
    print("n=== Evaluation Results ===")
    print(f"Final Status: {'Passed' if result['passed'] else 'Failed'} after {result['iteration']} iteration(s)")
    print(f"Final Code:n{result['code']}")
    print(f"Final Feedback:n{result['feedback']}")
    print("nIteration History:")
    for attempt in result["history"]:
        print(f"Iteration {attempt['iteration']}:")
        print(f"  Code:n{attempt['code']}")
        print(f"  Feedback: {attempt['feedback']}")
        print(f"  Passed: {attempt['passed']}n")


if __name__ == "__main__":
    main()

输出：

Iteration 1 - Code Writer: Generating code
Iteration 1 - Code Writer: Received feedback:
Iteration 1 - Code Writer: Code generated
Iteration 1 - Code Tester: Testing code
Iteration 1 - Code Tester: Testing complete - Failed
Iteration 1 - Loop continues: Tests failed
Iteration 2 - Code Writer: Generating code
Iteration 2 - Code Writer: Received feedback: Test failed: factorial(-1) should raise an error.
Iteration 2 - Code Writer: Code generated
Iteration 2 - Code Tester: Testing code
Iteration 2 - Code Tester: Testing complete - Passed
Iteration 2 - Loop stops: Tests passed
=== Evaluation Results ===
Final Status: Passed after 2 iteration(s)
Final Code:
def factorial(n):
    if n < 0:
        raise ValueError("Factorial not defined for negative numbers")
    if n == 0:
        return 1
    result = 1
    for i in range(1, n + 1):
        result *= i
    return result
Final Feedback:
All tests passed!
Iteration History:
Iteration 1:
  Code:
def factorial(n):
    result = 1
    for i in range(1, n + 1):
        result *= i
    return result
  Feedback: Test failed: factorial(-1) should raise an error.
  Passed: False
Iteration 2:
  Code:
def factorial(n):
    if n < 0:
        raise ValueError("Factorial not defined for negative numbers")
    if n == 0:
        return 1
    result = 1
    for i in range(1, n + 1):
        result *= i
    return result
  Feedback: All tests passed!
  Passed: True

综合反馈：代码测试器现在会报告所有测试失败的情况，确保代码编写器拥有逐步修复问题所需的信息。

正确的反馈处理：代码编写器优先处理修复（先处理零的情况，然后是负输入的情况），确保逐步改进。

循环终止：当测试通过时，循环会正确退出，而不会不必要地运行全部3次迭代。

2.1.4 路由器

一个中央路由器根据任务或输入来决定调用哪些代理。

示例：客户支持工单路由。

代码：

from typing import Dict, Any, TypedDict, Literal
from langgraph.graph import StateGraph, END
from langchain_core.runnables import RunnableConfig
import re
import time


class TicketState(TypedDict):
    ticket_text: str
    category: str
    resolution: str
    processing_time: float


def router_agent(state: TicketState) -> Dict[str, Any]:
    print("Router Agent: Analyzing ticket...")
    start_time = time.time()
    ticket_text = state["ticket_text"].lower()
    if any(keyword in ticket_text for keyword in ["billing", "payment", "invoice", "charge"]):
        category = "Billing"
    elif any(keyword in ticket_text for keyword in ["technical", "bug", "error", "crash"]):
        category = "Technical"
    elif any(keyword in ticket_text for keyword in ["general", "question", "inquiry", "info"]):
        category = "General"
    else:
        category = "Unknown"
    processing_time = time.time() - start_time
    print(f"Router Agent: Categorized as '{category}' in {processing_time:.2f} seconds")
    return {
        "category": category,
        "processing_time": processing_time
    }


def billing_team_agent(state: TicketState) -> Dict[str, Any]:
    print("Billing Team Agent: Processing ticket...")
    start_time = time.time()
    ticket_text = state["ticket_text"]
    resolution = f"Billing Team: Reviewed ticket '{ticket_text}'. Please check your invoice details or contact our billing department for further assistance."
    processing_time = time.time() - start_time
    time.sleep(1)
    print(f"Billing Team Agent: Completed in {processing_time:.2f} seconds")
    return {
        "resolution": resolution,
        "processing_time": state["processing_time"] + processing_time
    }


def technical_team_agent(state: TicketState) -> Dict[str, Any]:
    print("Technical Team Agent: Processing ticket...")
    start_time = time.time()
    ticket_text = state["ticket_text"]
    resolution = f"Technical Team: Reviewed ticket '{ticket_text}'. Please try restarting your device or submit a detailed error log for further investigation."
    processing_time = time.time() - start_time
    time.sleep(1.5)
    print(f"Technical Team Agent: Completed in {processing_time:.2f} seconds")
    return {
        "resolution": resolution,
        "processing_time": state["processing_time"] + processing_time
    }


def general_team_agent(state: TicketState) -> Dict[str, Any]:
    print("General Team Agent: Processing ticket...")
    start_time = time.time()
    ticket_text = state["ticket_text"]
    resolution = f"General Team: Reviewed ticket '{ticket_text}'. For more information, please refer to our FAQ or contact us via email."
    processing_time = time.time() - start_time
    time.sleep(0.8)
    print(f"General Team Agent: Completed in {processing_time:.2f} seconds")
    return {
        "resolution": resolution,
        "processing_time": state["processing_time"] + processing_time
    }


def manual_review_agent(state: TicketState) -> Dict[str, Any]:
    print("Manual Review Agent: Processing ticket...")
    start_time = time.time()
    ticket_text = state["ticket_text"]
    resolution = f"Manual Review: Ticket '{ticket_text}' could not be categorized. Flagged for human review. Please assign to the appropriate team manually."
    processing_time = time.time() - start_time
    time.sleep(0.5)
    print(f"Manual Review Agent: Completed in {processing_time:.2f} seconds")
    return {
        "resolution": resolution,
        "processing_time": state["processing_time"] + processing_time
    }


def route_ticket(state: TicketState) -> Literal["billing_team", "technical_team", "general_team", "manual_review"]:
    category = state["category"]
    print(f"Routing: Ticket category is '{category}'")
    if category == "Billing":
        return"billing_team"
    elif category == "Technical":
        return"technical_team"
    elif category == "General":
        return"general_team"
    else:
        return"manual_review"


def build_router_graph() -> StateGraph:
    workflow = StateGraph(TicketState)
    workflow.add_node("router", router_agent)
    workflow.add_node("billing_team", billing_team_agent)
    workflow.add_node("technical_team", technical_team_agent)
    workflow.add_node("general_team", general_team_agent)
    workflow.add_node("manual_review", manual_review_agent)
    workflow.set_entry_point("router")
    workflow.add_conditional_edges(
        "router",
        route_ticket,
        "billing_team": "billing_team",
        "technical_team": "technical_team",
        "general_team": "general_team",
        "manual_review": "manual_review"
    )
    workflow.add_edge("billing_team", END)
    workflow.add_edge("technical_team", END)
    workflow.add_edge("general_team", END)
    workflow.add_edge("manual_review", END)
    return workflow.compile()


def main():
    test_tickets = [
        "I have a billing issue with my last invoice. It seems I was overcharged.",
        "My app keeps crashing with a technical error. Please help!",
        "I have a general question about your services. Can you provide more info?",
        "I need assistance with something unrelated to billing or technical issues."
    ]
    for ticket_text in test_tickets:
        initial_state: TicketState = {
            "ticket_text": ticket_text,
            "category": "",
            "resolution": "",
            "processing_time": 0.0
        }
        print(f"n=== Processing Ticket: '{ticket_text}' ===")
        app = build_router_graph()
        start_time = time.time()
        result = app.invoke(initial_state, config=RunnableConfig())
        total_time = time.time() - start_time
        print("n=== Ticket Results ===")
        print(f"Category: {result['category']}")
        print(f"Resolution: {result['resolution']}")
        print(f"Total Processing Time: {result['processing_time']:.2f} seconds")
        print(f"Total Wall Clock Time: {total_time:.2f} seconds")
        print("-" * 50)


if __name__ == "__main__":
    main()