1
# 普通 LLM 调用
2
def普通调用(user_input):
3
    response = llm.generate(user_input)
4
    return response
5

6
# Agent 调用
7
def agent调用(user_input):
8
    # 1. 理解任务
9
    plan = llm.plan(user_input)
10

11
    # 2. 执行计划
12
    for step in plan.steps:
13
        if step.needs_tool:
14
            result = execute_tool(step.tool, step.args)
15
            plan.add_context(result)
16
        else:
17
            result = llm.reason(step.task)
18

19
    # 3. 返回结果
20
    return plan.final_answer

1
from openai import OpenAI
2

3
client = OpenAI()
4

5
# 定义工具 Schema
6
tools = [
7
    {
8
        "type": "function",
9
        "function": {
10
            "name": "search_code",
11
            "description": "搜索代码库中的代码",
12
            "parameters": {
13
                "type": "object",
14
                "properties": {
15
                    "query": {
16
                        "type": "string",
17
                        "description": "搜索关键词"
18
                    },
19
                    "language": {
20
                        "type": "string",
21
                        "description": "编程语言"
22
                    }
23
                },
24
                "required": ["query"]
25
            }
26
        }
27
    },
28
    {
29
        "type": "function",
30
        "function": {
31
            "name": "read_file",
32
            "description": "读取文件内容",
33
            "parameters": {
34
                "type": "object",
35
                "properties": {
36
                    "path": {
37
                        "type": "string",
38
                        "description": "文件路径"
39
                    },
40
                    "line_start": {
41
                        "type": "integer",
42
                        "description": "起始行号"
43
                    },
44
                    "line_end": {
45
                        "type": "integer",
46
                        "description": "结束行号"
47
                    }
48
                },
49
                "required": ["path"]
50
            }
51
        }
52
    }
53
]
54

55
# Agent 执行
56
def agent_execute(query):
57
    messages = [{"role": "user", "content": query}]
58

59
    while True:
60
        response = client.chat.completions.create(
61
            model="gpt-4-turbo",
62
            messages=messages,
63
            tools=tools,
64
            tool_choice="auto"
65
        )
66

67
        message = response.choices[0].message
68

69
        if message.tool_calls:
70
            # LLM 请求调用工具
71
            for tool_call in message.tool_calls:
72
                tool_name = tool_call.function.name
73
                tool_args = json.loads(tool_call.function.arguments)
74

75
                # 执行工具
76
                result = execute_tool(tool_name, tool_args)
77

78
                # 添加工具结果到上下文
79
                messages.append({
80
                    "role": "tool",
81
                    "tool_call_id": tool_call.id,
82
                    "content": result
83
                })
84
        else:
85
            # LLM 直接回答
86
            return message.content
87

88
def execute_tool(name, args):
89
    if name == "search_code":
90
        return search_code(**args)
91
    elif name == "read_file":
92
        return read_file(**args)

1
# 好的工具设计
2
good_tools = [
3
    {
4
        "name": "get_weather",
5
        "description": "获取指定城市的天气预报",
6
        "parameters": {
7
            "city": {"type": "string", "description": "城市名称"},
8
            "date": {"type": "string", "description": "日期，格式 YYYY-MM-DD"}
9
        },
10
        # 具体说明输入输出
11
    },
12
    {
13
        "name": "search_database",
14
        "description": "在数据库中搜索用户信息",
15
        "parameters": {
16
            "table": {"type": "string", "description": "表名"},
17
            "conditions": {"type": "object", "description": "查询条件"}
18
        },
19
        # 参数结构清晰
20
    }
21
]
22

23
# 不好的工具设计
24
bad_tools = [
25
    {
26
        "name": "query",
27
        "description": "执行查询",
28
        # 描述模糊
29
    },
30
    {
31
        "name": "do_something",
32
        "description": "做某事",
33
        # 名称无意义
34
    }
35
]

1
class ReActAgent:
2
    def __init__(self, llm, tools, max_iterations=10):
3
        self.llm = llm
4
        self.tools = tools
5
        self.max_iterations = max_iterations
6

7
    def run(self, task: str) -> str:
8
        """ReAct 主循环"""
9
        observations = []
10

11
        for i in range(self.max_iterations):
12
            # 1. Thought: LLM 分析当前状态，决定下一步
13
            thought = self.llm.think(
14
                task=task,
15
                history=observations,
16
                tools=self.tools
17
            )
18

19
            if thought.is_finished:
20
                return thought.answer
21

22
            # 2. Action: 执行工具调用
23
            if thought.needs_tool:
24
                action_result = self.execute_tool(
25
                    thought.tool_name,
26
                    thought.tool_args
27
                )
28
                observations.append({
29
                    "thought": thought,
30
                    "action": thought.action,
31
                    "observation": action_result
32
                })
33
            else:
34
                # 直接推理回答
35
                return thought.answer
36

37
        return "任务未在限制次数内完成"
38

39
    def execute_tool(self, tool_name, args):
40
        """执行工具"""
41
        for tool in self.tools:
42
            if tool.name == tool_name:
43
                return tool.execute(**args)
44
        raise ValueError(f"Unknown tool: {tool_name}")

1
REACT_PROMPT = """你是一个 AI Agent，需要完成用户任务。
2

3
可用工具：
4
{tool_schemas}
5

6
执行流程：
7
1. Thought: 分析当前情况，决定是否需要调用工具
8
2. Action: 如果需要，调用工具并传入参数
9
3. Observation: 观察工具返回结果
10
4. 重复直到任务完成
11

12
当前任务：{task}
13

14
{history}
15

16
请开始执行：
17
"""
18

19
# 带历史记录的 ReAct
20
def format_history(observations):
21
    history = ""
22
    for i, obs in enumerate(observations):
23
        history += f"步骤 {i+1}:\n"
24
        history += f"思考：{obs['thought']}\n"
25
        history += f"行动：{obs['action']}\n"
26
        history += f"观察：{obs['observation']}\n\n"
27
    return history

1
class PlanAndExecuteAgent:
2
    """
3
    计划-执行分离模式：
4
    1. 先生成完整执行计划
5
    2. 然后逐个执行计划中的步骤
6
    """
7

8
    def plan(self, task: str) -> list:
9
        """生成执行计划"""
10
        prompt = f"""分析以下任务，分解为具体执行步骤：
11

12
任务：{task}
13

14
请列出执行步骤（每步一个具体行动）：
15
"""
16
        steps = self.llm.generate(prompt)
17
        return self.parse_steps(steps)
18

19
    def execute(self, task: str):
20
        """执行计划"""
21
        # 1. 生成计划
22
        plan = self.plan(task)
23

24
        # 2. 逐个执行
25
        context = {}
26
        for i, step in enumerate(plan):
27
            print(f"执行步骤 {i+1}: {step}")
28

29
            # 根据上下文执行
30
            result = self.execute_step(step, context)
31
            context[step] = result
32

33
        return context
34

35
    def execute_step(self, step, context):
36
        """执行单个步骤"""
37
        if "search" in step:
38
            return self.search(context)
39
        elif "read" in step:
40
            return self.read_file(context)
41
        else:
42
            return self.llm.reason(step, context)

1
class TreeOfThought:
2
    def __init__(self, llm, num_thoughts=3, max_depth=3):
3
        self.llm = llm
4
        self.num_thoughts = num_thoughts
5
        self.max_depth = max_depth
6

7
    def solve(self, problem: str) -> str:
8
        """ToT 求解"""
9
        root = ThoughtNode(problem, depth=0)
10

11
        # BFS 探索
12
        queue = [root]
13
        best_solution = None
14
        best_score = -1
15

16
        while queue:
17
            node = queue.pop(0)
18

19
            if node.depth >= self.max_depth:
20
                # 评估叶节点
21
                score = self.evaluate(node)
22
                if score > best_score:
23
                    best_score = score
24
                    best_solution = node
25
                continue
26

27
            # 生成多个候选
28
            candidates = self.generate_candidates(node)
29

30
            for candidate in candidates:
31
                queue.append(candidate)
32

33
        return best_solution.path
34

35
    def generate_candidates(self, node) -> list:
36
        """生成候选子节点"""
37
        prompt = f"""当前问题：{node.problem}
38
当前方案：{node.solution}
39

40
请提出 {self.num_thoughts} 种不同的改进方案或下一步行动：
41
"""
42
        proposals = self.llm.generate(prompt)
43
        return [ThoughtNode(node, p) for p in proposals]
44

45
    def evaluate(self, node) -> float:
46
        """评估节点质量"""
47
        prompt = f"""评估以下方案的质量（0-10分）：
48

49
方案：{node.solution}
50

51
评分标准：
52
- 可行性（0-3分）
53
- 完整性（0-3分）
54
- 创新性（0-4分）
55

56
评分："""
57
        return self.llm.generate(prompt)

1
class SelfDiscoverAgent:
2
    """
3
    Self-Discover: Agent 自己发现解决任务的最佳策略
4
    """
5

6
    def __init__(self, llm):
7
        self.llm = llm
8

9
    def discover(self, task: str) -> str:
10
        # 1. 列出可能的策略
11
        strategies = self.list_strategies(task)
12

13
        # 2. 选择相关策略
14
        relevant = self.select_relevant(strategies, task)
15

16
        # 3. 组合为执行计划
17
        plan = self.compose_plan(relevant, task)
18

19
        # 4. 执行
20
        return self.execute(plan)
21

22
    def list_strategies(self, task):
23
        prompt = f"""对于以下任务，列出所有可能的解决策略：
24

25
任务：{task}
26

27
策略类型：
28
1. 分解为子问题
29
2. 类比推理
30
3. 逐步验证
31
4. 反向思考
32
5. 广度优先 vs 深度优先
33
"""
34
        return self.llm.generate(prompt)

1
class AgentMemory:
2
    def __init__(self):
3
        # 短期记忆：最近对话
4
        self.short_term = deque(maxlen=20)
5

6
        # 长期记忆：向量数据库
7
        self.long_term = VectorStore()
8

9
        # 工作记忆：当前上下文
10
        self.working_memory = {}
11

12
    def add_turn(self, role: str, content: str):
13
        """添加对话"""
14
        self.short_term.append({"role": role, "content": content})
15

16
    def add_experience(self, experience: dict):
17
        """添加经验到长期记忆"""
18
        vector = self.embed(experience["content"])
19
        self.long_term.add(
20
            id=experience["id"],
21
            vector=vector,
22
            content=experience["content"],
23
            metadata=experience.get("metadata", {})
24
        )
25

26
    def retrieve(self, query: str, top_k: int = 3) -> list:
27
        """检索相关记忆"""
28
        query_vector = self.embed(query)
29
        return self.long_term.search(query_vector, top_k)
30

31
    def summarize_long_term(self):
32
        """定期总结长期记忆，提取重要信息"""
33
        all_memories = self.long_term.get_all()
34

35
        prompt = f"""总结以下经验，提取关键模式：
36

37
{all_memories}
38

39
总结：
40
"""
41
        summary = self.llm.generate(prompt)
42
        return summary

1
class MemoryAugmentedRetrieval:
2
    def __init__(self, memory, retriever):
3
        self.memory = memory
4
        self.retriever = retriever
5

6
    def retrieve_with_memory(self, query: str, context: dict) -> list:
7
        """结合记忆的检索"""
8
        # 1. 基础检索
9
        base_results = self.retriever.search(query, top_k=5)
10

11
        # 2. 相关记忆检索
12
        memory_results = self.memory.retrieve(query, top_k=3)
13

14
        # 3. 融合结果
15
        fused = self._fusion_results(base_results, memory_results)
16

17
        # 4. 重排序
18
        reranked = self._rerank(query, fused, context)
19

20
        return reranked
21

22
    def _fusion_results(self, base, memory):
23
        """融合检索结果"""
24
        scores = {}
25

26
        for i, item in enumerate(base):
27
            scores[item.id] = scores.get(item.id, 0) + 1 / (i + 1)
28

29
        for i, item in enumerate(memory):
30
            scores[item.id] = scores.get(item.id, 0) + 1 / (i + 1) * 0.5
31

32
        return sorted(scores.items(), key=lambda x: -x[1])

1
from langchain.agents import AgentExecutor, create_react_agent
2
from langchain.tools import Tool
3
from langchain_openai import ChatOpenAI
4

5
# 定义工具
6
tools = [
7
    Tool(
8
        name="search",
9
        func=search_code,
10
        description="搜索代码库中的代码"
11
    ),
12
    Tool(
13
        name="read",
14
        func=read_file,
15
        description="读取文件内容"
16
    )
17
]
18

19
# 创建 Agent
20
llm = ChatOpenAI(model="gpt-4-turbo")
21
agent = create_react_agent(llm, tools, prompt=REACT_PROMPT)
22

23
# 执行
24
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
25
result = agent_executor.invoke({"input": "查找处理用户认证的代码"})

1
from crewai import Agent, Task, Crew
2

3
# 定义 Agent（角色）
4
researcher = Agent(
5
    role="研究员",
6
    goal="收集相关信息",
7
    backstory="你是一个专业的研究员，擅长信息收集",
8
    tools=[search_tool, browse_tool]
9
)
10

11
writer = Agent(
12
    role="作家",
13
    goal="撰写报告",
14
    backstory="你是一个专业的技术作家",
15
    tools=[write_tool]
16
)
17

18
# 定义任务
19
research_task = Task(
20
    description="研究 AI Agent 的最新发展",
21
    agent=researcher
22
)
23

24
write_task = Task(
25
    description="撰写研究报告",
26
    agent=writer,
27
    context=[research_task]  # 依赖研究任务
28
)
29

30
# 创建 Crew
31
crew = Crew(agents=[researcher, writer], tasks=[research_task, write_task])
32
result = crew.kickoff()