盘点 10 个值得关注的 AI Agent Harness Engineering 开源项目

引言

在人工智能技术快速发展的今天，AI Agent（智能代理）正在成为下一代软件范式的核心。从简单的聊天机器人到复杂的自主决策系统，AI Agent 正在改变我们与计算机交互的方式。但是，构建一个真正强大、可靠且可扩展的 AI Agent 并不是一件容易的事情。

这就是为什么 AI Agent Harness Engineering（AI 代理框架工程） 如此重要的原因。它提供了一套标准化的工具、框架和最佳实践，帮助开发者更高效地构建、测试和部署 AI Agent。

在这篇文章中，我将带你深入了解 10 个最值得关注的 AI Agent Harness Engineering 开源项目。无论你是刚入门的新手，还是经验丰富的 AI 从业者，这篇文章都将为你提供有价值的参考。

目录

1. 核心概念解析：什么是 AI Agent Harness Engineering？
2. 项目评选标准与方法论
3. 10 大开源项目深度解析
4. 项目对比与选型指南
5. 实际应用场景与案例分析
6. 未来发展趋势与挑战
7. 总结与建议

---

核心概念解析

什么是 AI Agent？

在深入探讨框架之前，让我们先明确什么是 AI Agent。AI Agent 是一种能够感知环境、做出决策并采取行动的自主系统。它通常具备以下几个核心能力：

• 感知能力：获取和理解环境信息
• 推理能力：基于感知到的信息进行逻辑推理
• 决策能力：根据推理结果选择合适的行动
• 执行能力：实际执行所选的行动
• 学习能力：从经验中学习并改进自身行为

AI Agent Harness Engineering 的定义

AI Agent Harness Engineering 是指设计、构建和管理 AI Agent 基础设施的工程实践。它涵盖了 Agent 的生命周期管理、工具集成、安全控制、性能优化等多个方面。

简单来说，如果把 AI Agent 比作一辆汽车，那么 Harness Engineering 就是构建汽车的生产线、维修站和交通管理系统。

为什么需要 AI Agent Harness Engineering？

在 AI Agent 开发过程中，开发者通常会面临以下挑战：

1. 复杂性管理：AI Agent 通常涉及多个组件（LLM、工具、记忆系统等）的协调
2. 可观测性：难以追踪和理解 Agent 的决策过程
3. 安全性：确保 Agent 不会做出有害或不安全的行为
4. 可扩展性：如何让 Agent 处理更复杂的任务
5. 测试与调试：如何有效地测试和调试 Agent 的行为

AI Agent Harness Engineering 正是为了解决这些挑战而诞生的。

---

项目评选标准与方法论

在选择这 10 个开源项目时，我使用了以下评估标准：

评估维度	权重	说明
GitHub Stars	15%	社区受欢迎程度
活跃度	15%	最近更新频率、贡献者数量
文档质量	15%	文档完整性、示例丰富度
功能完整性	20%	核心功能覆盖度
架构设计	15%	可扩展性、模块化程度
社区生态	10%	插件生态、第三方集成
生产就绪度	10%	企业采用案例、稳定性

基于这些标准，我从上百个相关项目中筛选出了以下 10 个最具代表性的项目。

---

10 大开源项目深度解析

1. LangChain

项目概述

LangChain 无疑是目前最流行的 AI Agent 开发框架。它由 Harrison Chase 于 2022 年 10 月创建，旨在帮助开发者更容易地构建由语言模型驱动的应用程序。

GitHub: https://github.com/langchain-ai/langchain
Stars: ~80,000+
主要语言: Python, TypeScript
许可证: MIT

核心概念

LangChain 引入了几个关键概念：

1. Chains（链）：将多个组件组合在一起
2. Agents（代理）：使用 LLM 决定下一步行动
3. Tools（工具）：Agent 可以使用的外部功能
4. Memory（记忆）：在交互之间保持状态
5. Indexes（索引）：结构化文档以便检索

架构设计

核心代码示例

让我们看一个简单的 LangChain Agent 实现：

from langchain.agents import initialize_agent, Tool
from langchain.agents import AgentType
from langchain.llms import OpenAI
from langchain.tools import DuckDuckGoSearchRun

# 初始化语言模型
llm = OpenAI(temperature=0)

# 定义搜索工具
search = DuckDuckGoSearchRun()

tools = [
    Tool(
        name="Current Search",
        func=search.run,
        description="useful for when you need to answer questions about current events or the current state of the world"
    ),
]

# 初始化 agent
agent = initialize_agent(
    tools, 
    llm, 
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION, 
    verbose=True
)

# 使用 agent
agent.run("What's the latest news about AI development in 2024?")

优缺点分析

优点：

• 庞大的社区支持和丰富的文档
• 丰富的集成生态系统
• 多语言支持（Python、TypeScript）
• 持续快速迭代

缺点：

• 学习曲线较陡
• 某些抽象层次可能过于复杂
• 版本更新频繁，可能导致兼容性问题

---

2. AutoGPT

项目概述

AutoGPT 是一个实验性的开源 AI 代理，展示了 GPT-4 的能力。它由 Significant Gravitas 开发，能够自主地设定和完成目标。

GitHub: https://github.com/Significant-Gravitas/AutoGPT
Stars: ~150,000+
主要语言: Python
许可证: MIT

核心概念

AutoGPT 的核心创新在于其 自主循环（Autonomous Loop）：

1. 思考（Thoughts）：分析当前状态
2. 推理（Reasoning）：推导下一步行动
3. 计划（Plan）：制定具体策略
4. 批评（Criticism）：自我反思和评估
5. 执行（Execute）：采取行动

工作原理

核心代码实现

让我们看一下 AutoGPT 的核心循环实现：

class Agent:
    def __init__(self, ai_name, ai_role, ai_goals, memory):
        self.ai_name = ai_name
        self.ai_role = ai_role
        self.ai_goals = ai_goals
        self.memory = memory
        self.completed_tasks = []
        
    def run(self):
        """运行自主代理的主循环"""
        loop_count = 0
        max_loops = 100
        
        while loop_count < max_loops:
            loop_count += 1
            
            # 1. 构建提示词
            prompt = self._construct_prompt()
            
            # 2. 与 LLM 交互获取思维和动作
            thoughts, actions = self._get_llm_response(prompt)
            
            # 3. 展示思考过程
            self._display_thoughts(thoughts)
            
            # 4. 执行动作
            result = self._execute_actions(actions)
            
            # 5. 存储到记忆
            self.memory.add(f"Round {loop_count}: {thoughts} -> {actions} -> {result}")
            
            # 6. 检查是否完成
            if self._check_goal_completion(result):
                print(f"目标已完成！共进行 {loop_count} 轮迭代。")
                return result
                
        print("已达到最大迭代次数，任务未完成。")
        return None
        
    def _construct_prompt(self):
        """构建发送给 LLM 的提示词"""
        context = self.memory.get_recent(5)  # 获取最近5条记忆
        
        return f"""
你是 {self.ai_name}，你的角色是 {self.ai_role}。

你的目标：
{self._format_goals()}

已完成的任务：
{self._format_completed_tasks()}

最近的上下文：
{self._format_context(context)}

请按照以下格式响应：
THOUGHTS: <你的思考>
REASONING: <你的推理>
PLAN: <你的计划>
CRITICISM: <自我批评>
NEXT ACTION: <要执行的动作>
"""

实际应用场景

• 研究助手：自动搜索和总结研究论文
• 代码开发：根据需求自动编写和调试代码
• 内容创作：自动生成文章、报告和营销文案
• 个人助理：管理日程、处理邮件、安排会议

---

3. BabyAGI

项目概述

BabyAGI 是另一个开创性的 AI 代理项目，由 Yohei Nakajima 创建。它的核心理念是创建一个能够自我指导完成目标的任务管理系统。

GitHub: https://github.com/yoheinakajima/babyagi
Stars: ~20,000+
主要语言: Python
许可证: MIT

核心原理

BabyAGI 的工作流程基于三个主要组件：

1. 任务创建代理（Task Creation Agent）：根据目标和已完成任务创建新任务
2. 任务优先级代理（Prioritization Agent）：重新排列任务优先级
3. 执行代理（Execution Agent）：执行优先级最高的任务

数学模型

BabyAGI 使用向量相似度来检索相关上下文，其核心公式是余弦相似度：

$$\text{similarity}(A,B)=\frac{A\cdot B}{\Vert A\Vert \Vert B\Vert }=\frac{{\sum }_{i=1}^n{A}_i{B}_i}{\sqrt{{\sum }_{i=1}^n{A}_i^2}\sqrt{{\sum }_{i=1}^n{B}_i^2}}$$

这个公式用于计算任务描述与存储记忆之间的相似度，从而找到最相关的上下文。

核心流程

代码实现

下面是 BabyAGI 的核心实现：

import openai
import pinecone
from collections import deque
from typing import Dict, List

class BabyAGI:
    def __init__(self, openai_api_key, pinecone_api_key, pinecone_environment, table_name):
        # 初始化 API 客户端
        openai.api_key = openai_api_key
        pinecone.init(api_key=pinecone_api_key, environment=pinecone_environment)
        
        # 连接到 Pinecone 索引
        self.index = pinecone.Index(table_name)
        
        # 任务队列
        self.task_list = deque()
        
    def add_task(self, task: Dict):
        """添加任务到队列"""
        self.task_list.append(task)
        
    def get_embedding(self, text: str) -> List[float]:
        """获取文本的向量嵌入"""
        response = openai.Embedding.create(
            input=text,
            model="text-embedding-ada-002"
        )
        return response['data'][0]['embedding']
        
    def context_agent(self, query: str, top_k: int = 5) -> List[str]:
        """根据查询检索相关上下文"""
        query_embedding = self.get_embedding(query)
        
        results = self.index.query(
            vector=query_embedding,
            top_k=top_k,
            include_metadata=True
        )
        
        # 返回相关的任务结果
        return [item['metadata']['task'] + ": " + item['metadata']['result'] 
                for item in results['matches']]
                
    def execution_agent(self, objective: str, task: str) -> str:
        """执行代理：执行给定的任务"""
        context = self.context_agent(query=objective, top_k=5)
        context_str = "\n".join(context)
        
        prompt = f"""你是一个执行具体任务的AI助手。
你的终极目标是：{objective}

以下是已完成任务的上下文：
{context_str}

现在请执行以下任务：{task}
请提供详细的执行结果。"""

        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages=[{"role": "user", "content": prompt}]
        )
        
        return response.choices[0].message.content
        
    def task_creation_agent(self, objective: str, result: str, task_description: str, task_list: List[str]) -> List[Dict]:
        """任务创建代理：基于已完成的任务创建新任务"""
        prompt = f"""你是一个任务创建AI助手。
你的终极目标是：{objective}

刚刚完成的任务：{task_description}
任务执行结果：{result}

当前待处理任务列表：{task_list}

请基于以上信息，判断是否需要创建新的任务。
如果需要，请以JSON数组格式返回新任务，格式如下：
[
  {{"task_name": "任务1描述"}},
  {{"task_name": "任务2描述"}}
]
如果不需要创建新任务，请返回空数组[]。"""

        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages=[{"role": "user", "content": prompt}]
        )
        
        try:
            return eval(response.choices[0].message.content)
        except:
            return []
            
    def prioritization_agent(self, objective: str, task_list: List[str]) -> List[str]:
        """优先级代理：重新排列任务优先级"""
        task_names = [t["task_name"] for t in task_list]
        next_task_id = len(task_list) + 1
        
        prompt = f"""你是一个任务优先级排序AI助手。
你的终极目标是：{objective}

当前任务列表：
{task_names}

请按照重要性和逻辑顺序重新排列这些任务。
只返回任务名称列表，格式如下：
["任务1", "任务2", "任务3"]"""

        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages=[{"role": "user", "content": prompt}]
        )
        
        try:
            new_tasks = eval(response.choices[0].message.content)
            return [{"task_id": next_task_id + i, "task_name": task_name} 
                    for i, task_name in enumerate(new_tasks)]
        except:
            return task_list
            
    def run(self, objective: str, first_task: str):
        """运行 BabyAGI 的主循环"""
        # 添加第一个任务
        self.add_task({"task_id": 1, "task_name": first_task})
        
        task_id_counter = 1
        
        while True:
            if self.task_list:
                # 1. 打印当前任务列表
                print("\033[95m\033[1m" + "\n*****TASK LIST*****\n" + "\033[0m\033[0m")
                for t in self.task_list:
                    print(str(t['task_id']) + ": " + t['task_name'])
                    
                # 2. 取出第一个任务
                task = self.task_list.popleft()
                print("\033[92m\033[1m" + "\n*****NEXT TASK*****\n" + "\033[0m\033[0m")
                print(str(task['task_id']) + ": " + task['task_name'])
                
                # 3. 执行任务
                result = self.execution_agent(objective, task['task_name'])
                print("\033[93m\033[1m" + "\n*****TASK RESULT*****\n" + "\033[0m\033[0m")
                print(result)
                
                # 4. 存储结果
                enriched_result = {"data": result}
                result_id = f"result_{task['task_id']}"
                vector = self.get_embedding(enriched_result['data'])
                self.index.upsert([(result_id, vector, {"task": task['task_name'], "result": result})])
                
                # 5. 创建新任务
                new_tasks = self.task_creation_agent(
                    objective, 
                    result, 
                    task['task_name'], 
                    [t['task_name'] for t in self.task_list]
                )
                
                for new_task in new_tasks:
                    task_id_counter += 1
                    new_task.update({"task_id": task_id_counter})
                    self.add_task(new_task)
                    
                # 6. 重新排列任务优先级
                self.task_list = deque(self.prioritization_agent(objective, list(self.task_list)))

---

4. LangGraph

项目概述

LangGraph 是 LangChain 团队推出的一个用于构建有状态代理的库。它是 LangChain 生态系统的一部分，专门设计用来解决更复杂的代理工作流问题。

GitHub: https://github.com/langchain-ai/langgraph
Stars: ~3,000+
主要语言: Python
许可证: MIT

核心概念

LangGraph 引入了几个关键概念：

1. StateGraph（状态图）：定义代理的状态和转换
2. Nodes（节点）：表示代理可以执行的动作或计算
3. Edges（边）：连接节点，定义控制流
4. Conditional Edges（条件边）：根据当前状态决定下一步
5. State（状态）：在节点之间传递的数据

为什么需要 LangGraph？

虽然 LangChain 提供了构建代理的基础组件，但对于复杂的、有状态的工作流，传统的 Chain 模式显得力不从心。LangGraph 正是为了解决这个问题：

• 循环能力：支持自然的循环和条件分支
• 状态管理：内置状态传递和管理机制
• 可控性：更精细地控制代理的执行流程
• 可观测性：更好的调试和可视化工具

架构设计

代码示例

让我们看一个使用 LangGraph 构建的简单代理：

from typing import Annotated, Literal, TypedDict

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

# 1. 定义工具
@tool
def search(query: str) -> str:
    """搜索网络获取最新信息"""
    # 在实际应用中，这里会连接到真实的搜索 API
    return f"搜索结果：关于 '{query}' 的最新信息显示..."

@tool
def calculate(expression: str) -> str:
    """计算数学表达式"""
    try:
        result = eval(expression)
        return f"计算结果：{expression} = {result}"
    except:
        return "计算错误：无效的表达式"

tools = [search, calculate]

# 2. 定义状态
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]

# 3. 初始化模型和工具节点
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
llm_with_tools = llm.bind_tools(tools)
tool_node = ToolNode(tools)

# 4. 定义节点函数
def agent_node(state: AgentState):
    messages = state["messages"]
    response = llm_with_tools.invoke(messages)
    return {"messages": [response]}

def should_continue(state: AgentState) -> Literal["tools", END]:
    """决定是继续使用工具还是结束"""
    messages = state["messages"]
    last_message = messages[-1]
    if last_message.tool_calls:
        return "tools"
    return END

# 5. 构建图
workflow = StateGraph(AgentState)

# 添加节点
workflow.add_node("agent", agent_node)
workflow.add_node("tools", tool_node)

# 添加边
workflow.add_edge(START, "agent")
workflow.add_conditional_edges("agent", should_continue)
workflow.add_edge("tools", "agent")

# 编译图
app = workflow.compile()

# 6. 使用代理
query = "2024年AI领域的重大突破是什么？请计算2024+2025等于多少？"
result = app.invoke({"messages": [("user", query)]})
print(result["messages"][-1].content)

高级特性：持久化和人类干预

LangGraph 还提供了一些高级特性，使得构建生产级代理变得更加容易：

from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.prebuilt import create_react_agent

# 使用 SQLite 持久化状态
memory = SqliteSaver.from_conn_string(":memory:")

# 创建带有人类干预的代理
app = create_react_agent(
    llm, 
    tools, 
    checkpointer=memory,
    interrupt_before=["tools"]  # 在使用工具前中断，等待人类确认
)

# 运行代理
config = {"configurable": {"thread_id": "1"}}
query = "搜索2024年最新的AI发展趋势"

# 第一次调用，会在使用工具前中断
result = app.invoke({"messages": [("user", query)]}, config)
print("代理想要使用工具，请确认：")
print(result["messages"][-1].tool_calls)

# 人类确认后继续执行
result = app.invoke(None, config)
print("\n最终结果：")
print(result["messages"][-1].content)

---

5. CrewAI

项目概述

CrewAI 是一个专注于多代理协作的框架。它的核心理念是让多个 AI 代理像团队一样协同工作，每个代理有自己的角色、目标和工具。

GitHub: https://github.com/joaomdmoura/crewAI
Stars: ~15,000+
主要语言: Python
许可证: MIT

核心概念

CrewAI 引入了一套独特的概念：

1. Agents（代理）：有特定角色和目标的个体
2. Tasks（任务）：分配给代理的具体工作
3. Crews（团队）：协同工作的代理组合
4. Processes（流程）：定义团队如何协作
5. Tools（工具）：代理可以使用的工具

代理设计原则

CrewAI 的代理设计遵循以下原则：

• 角色定位：每个代理有明确的角色定义
• 目标驱动：每个代理有自己的具体目标
• 背景故事：赋予代理个性和背景
• 自主权：代理可以自主决策如何完成任务
• 协作能力：代理可以询问其他代理和委派任务

架构设计

代码实现

让我们看一个使用 CrewAI 构建软件开发团队的例子：

from crewai import Agent, Task, Crew, Process
from langchain_openai import ChatOpenAI
from langchain.tools import DuckDuckGoSearchRun, tool

# 1. 初始化工具
search = DuckDuckGoSearchRun()

@tool
def write_file(filename: str, content: str) -> str:
    """写入内容到文件"""
    with open(filename, 'w') as f:
        f.write(content)
    return f"文件 {filename} 写入成功"

@tool
def read_file(filename: str) -> str:
    """读取文件内容"""
    try:
        with open(filename, 'r') as f:
            return f.read()
    except:
        return f"文件 {filename} 不存在"

tools = [search, write_file, read_file]

# 2. 初始化大语言模型
llm = ChatOpenAI(model="gpt-4", temperature=0.7)

# 3. 创建代理
product_manager = Agent(
    role='产品经理',
    goal='确保开发出满足用户需求的优质产品',
    backstory="""你是一位经验丰富的产品经理，擅长理解用户需求，
    并将其转化为清晰的产品规格。你善于平衡用户体验、技术可行性和业务目标。""",
    verbose=True,
    allow_delegation=True,
    tools=[search],
    llm=llm
)

tech_architect = Agent(
    role='技术架构师',
    goal='设计健壮、可扩展的系统架构',
    backstory="""你是一位资深的技术架构师，拥有丰富的系统设计经验。
    你擅长将复杂的问题分解为可管理的组件，并选择最合适的技术栈。""",
    verbose=True,
    allow_delegation=True,
    tools=[search],
    llm=llm
)

developer = Agent(
    role='高级开发者',
    goal='编写高质量、可维护的代码',
    backstory="""你是一位技术精湛的高级开发者，对多种编程语言和框架都有深入了解。
    你注重代码质量、可测试性和最佳实践。""",
    verbose=True,
    allow_delegation=False,
    tools=[search, write_file, read_file],
    llm=llm
)

tester = Agent(
    role='质量保证工程师',
    goal='确保产品质量和用户体验',
    backstory="""你是一位细心的质量保证工程师，善于发现软件中的缺陷和改进点。
    你注重细节，对产品质量有高标准。""",
    verbose=True,
    allow_delegation=True,
    tools=[search],
    llm=llm
)

# 4. 创建任务
task_define_requirements = Task(
    description="""分析并定义一个简单的待办事项应用的产品需求。
    需要考虑以下方面：
    1. 核心功能需求
    2. 用户界面需求
    3. 非功能性需求（性能、安全等）
    请在网上搜索类似产品的最佳实践，并形成一份产品需求文档大纲。""",
    agent=product_manager,
    expected_output="一份详细的产品需求文档大纲"
)

task_design_architecture = Task(
    description="""基于产品需求，设计待办事项应用的技术架构。
    需要考虑：
    1. 整体架构风格（如 MVC, REST API 等）
    2. 技术栈选择
    3. 数据库设计
    4. API 设计
    请提供详细的架构设计文档。""",
    agent=tech_architect,
    expected_output="一份详细的技术架构设计文档"
)

task_implement_code = Task(
    description="""基于技术架构设计，实现待办事项应用的核心代码。
    要求：
    1. 使用 Python 作为主要开发语言
    2. 实现核心的 CRUD 功能
    3. 包含基本的错误处理
    4. 代码要有适当的注释
    请将代码写入相应的文件中。""",
    agent=developer,
    expected_output="实现了待办事项应用核心功能的代码文件"
)

task_test_application = Task(
    description="""对待办事项应用进行全面的测试。
    需要包括：
    1. 功能测试
    2. 边界条件测试
    3. 错误处理测试
    请提供详细的测试报告，包括发现的问题和改进建议。""",
    agent=tester,
    expected_output="一份详细的测试报告"
)

# 5. 创建团队
crew = Crew(
    agents=[product_manager, tech_architect, developer, tester],
    tasks=[task_define_requirements, task_design_architecture, task_implement_code, task_test_application],
    process=Process.sequential,  # 按顺序执行任务
    verbose=2
)

# 6. 启动团队
result = crew.kickoff()
print("团队协作结果：")
print(result)

高级特性：流程和协作

CrewAI 支持不同的协作流程：

# 1. 顺序流程（Sequential）
crew_sequential = Crew(
    agents=[agent1, agent2, agent3],
    tasks=[task1, task2, task3],
    process=Process.sequential
)

# 2. 分层流程（Hierarchical） - 自动创建一个"经理"代理来协调任务
crew_hierarchical = Crew(
    agents=[agent1, agent2, agent3],
    tasks=[task1, task2, task3],
    process=Process.hierarchical,
    manager_llm=ChatOpenAI(model="gpt-4")
)

---

6. AutoGen

项目概述

AutoGen 是微软研究院推出的一个多代理框架，它允许开发者创建可定制的代理，并使它们能够相互协作来完成任务。

GitHub: https://github.com/microsoft/autogen
Stars: ~20,000+
主要语言: Python
许可证: MIT

核心概念

AutoGen 的核心概念包括：

1. Conversable Agents（可对话代理）：能够进行对话的代理
2. Assistant Agents（助手代理）：能够使用工具和生成代码的代理
3. User Proxy Agents（用户代理）：代表用户执行代码和获取输入的代理
4. Group Chat（群聊）：多个代理之间的协作对话
5. Agent Chat（代理对话）：两个代理之间的交互模式

设计理念

AutoGen 的设计遵循以下理念：

• 对话为中心：将代理交互建模为对话
• 可组合性：代理可以灵活组合
• 可扩展性：易于扩展和定制
• 人类参与：支持人类在回路中
• 工具使用：内置工具使用和代码执行能力

架构设计

代码示例：双代理对话

让我们看一个最基本的 AutoGen 示例：

from autogen import AssistantAgent, UserProxyAgent, config_list_from_json

# 1. 加载配置
config_list = config_list_from_json(env_or_file="OAI_CONFIG_LIST")

# 2. 创建助手代理
assistant = AssistantAgent(
    name="assistant",
    llm_config={
        "config_list": config_list,
        "temperature": 0
    }
)

# 3. 创建用户代理
user_proxy = UserProxyAgent(
    name="user_proxy",
    human_input_mode="NEVER",  # 不需要人类输入
    max_consecutive_auto_reply=10,
    is_termination_msg=lambda x: x.get("content", "").rstrip().endswith("TERMINATE"),
    code_execution_config={
        "work_dir": "coding",
        "use_docker": False  # 不使用 Docker，生产环境建议使用
    }
)

# 4. 开始对话
user_proxy.initiate_chat(
    assistant,
    message="""写一个 Python 脚本，计算斐波那契数列的前 20 项，并将结果保存到 fibonacci.txt 文件中。
请确保代码可以直接运行。完成后请说 TERMINATE。"""
)

代码示例：多代理群聊

现在让我们看一个更复杂的多代理协作示例：

from autogen import AssistantAgent, UserProxyAgent, GroupChat, GroupChatManager, config_list_from_json

# 1. 加载配置
config_list = config_list_from_json(env_or_file="OAI_CONFIG_LIST")
llm_config = {"config_list": config_list, "temperature": 0.7}

# 2. 创建不同角色的代理
user_proxy = UserProxyAgent(
    name="Admin",
    system_message="""你是管理员。你可以与团队讨论计划，但最终需要你批准计划。
    如果对计划满意，请回复：APPROVED。否则，回复修改建议。""",
    code_execution_config=False,
    human_input_mode="NEVER"
)

engineer = AssistantAgent(
    name="Engineer",
    llm_config=llm_config,
    system_message="""你是一位软件工程师。你遵循以下原则：
    1. 首先提出清晰简洁的计划，包括需要使用的文件列表
    2. 然后逐步实现代码，每次只实现一部分
    3. 你有写代码的权限，将代码保存到文件
    4. 以完整和可运行的代码结束
    """
)

scientist = AssistantAgent(
    name="Scientist",
    llm_config=llm_config,
    system_message="""你是一位研究科学家。你擅长进行研究和数据分析。
    你的职责是：
    1. 提供基于研究的见解
    2. 分析数据和解释结果
    3. 提出基于证据的建议
    """
)

planner = AssistantAgent(
    name="Planner",
    llm_config=llm_config,
    system_message="""你是一位计划专家。你擅长制定详细的执行计划。
    请确保计划是：
    1. 具体和可操作的
    2. 有明确的步骤和时间线
    3. 考虑到潜在的风险和缓解措施
    """
)

critic = AssistantAgent(
    name="Critic",
    llm_config=llm_config,
    system_message="""你是一位评论家。你擅长发现计划和想法中的潜在问题。
    请提供建设性的批评，关注：
    1. 逻辑一致性
    2. 可行性
    3. 潜在的风险
    4. 改进建议
    """
)

# 3. 创建群聊
groupchat = GroupChat(
    agents=[user_proxy, engineer, scientist, planner, critic],
    messages=[],
    max_round=20
)

manager = GroupChatManager(
    groupchat=groupchat,
    llm_config=llm_config
)

# 4. 开始对话
user_proxy.initiate_chat(
    manager,
    message="""我们需要构建一个简单的数据分析工具，用于分析销售数据。
    要求：
    1. 能够从 CSV 文件读取销售数据
    2. 进行基本的统计分析
    3. 生成可视化图表
    4. 生成分析报告
    
    请团队讨论并制定一个实现计划。"""
)

---

7. LlamaIndex (GPT Index)

项目概述

LlamaIndex（原名为 GPT Index）是一个用于连接 LLM 和私有数据的框架。虽然它不仅仅是一个代理框架，但它提供了构建数据感知代理的强大工具。

GitHub: https://github.com/run-llama/llama_index
Stars: ~30,000+
主要语言: Python
许可证: MIT

核心概念

LlamaIndex 的核心概念包括：

1. Data Connectors（数据连接器）：从不同来源摄入数据
2. Documents/Nodes（文档/节点）：表示数据的基本单元
3. Indices（索引）：组织数据以便高效检索
4. Retrievers（检索器）：从索引中获取相关数据
5. Query Engines（查询引擎）：端到端的问答流程
6. Agents（代理）：自动决定何时使用索引

索引类型

LlamaIndex 提供了多种索引类型，每种适用于不同的场景：

索引类型	描述	适用场景
Vector Store Index	基于向量相似度的索引	语义搜索、相似度匹配
List Index	简单的文档列表	小型数据集、顺序处理
Tree Index	层次化的树结构索引	长文档、需要摘要的场景
Keyword Table Index	基于关键词的索引	精确匹配、关键词搜索
Knowledge Graph Index	知识图谱索引	复杂关系推理、图查询

架构设计

代码示例：基本 RAG 系统

让我们看一个使用 LlamaIndex 构建基本 RAG（检索增强生成）系统的例子：

from llama_index import VectorStoreIndex, SimpleDirectoryReader, ServiceContext
from llama_index.llms import OpenAI

# 1. 加载文档
documents = SimpleDirectoryReader("data/").load_data()

# 2. 配置服务上下文（可以自定义 LLM、嵌入模型等）
llm = OpenAI(model="gpt-3.5-turbo", temperature=0)
service_context = ServiceContext.from_defaults(llm=llm)

# 3. 创建索引
index = VectorStoreIndex.from_documents(
    documents,
    service_context=service_context
)

# 4. 创建查询引擎
query_engine = index.as_query_engine()

# 5. 查询
response = query_engine.query("文档中讨论的主要观点是什么？")
print(response)

代码示例：构建数据代理

现在让我们看一个更高级的例子，构建一个能够使用多个工具的数据代理：

from llama_index import VectorStoreIndex, SimpleDirectoryReader, ServiceContext
from llama_index.llms import OpenAI
from llama_index.tools import QueryEngineTool, ToolMetadata
from llama_index.agent import OpenAIAgent

# 1. 加载不同的文档集
financial_docs = SimpleDirectoryReader("data/financial/").load_data()
product_docs = SimpleDirectoryReader("data/products/").load_data()
employee_docs = SimpleDirectoryReader("data/employees/").load_data()

# 2. 为每个文档集创建索引和查询引擎
service_context = ServiceContext.from_defaults(llm=OpenAI(model="