LLM 应用开发 - 从 Prompt Engineering 到 Agent 架构

#  אצל 好的提示
prompt = """你是一位经验丰富的 Go 语言工程师。
请用专业的方式回答以下问题，提供代码示例和最佳实践。
"""

# 不好的提示
prompt = """回答下面的问题。
"""

# 示例
question = "如何实现一个 Goroutine Pool？"
full_prompt = f"{prompt}\nQuestion: {question}"

# 引导模型逐步思考
prompt = """你是一位数学专家。请逐步推理解决问题。

问题: 小明有 3 个苹果，又买了 2 个，吃了 1 个，还剩几个？

步骤:
1. 初始数量: 3 个
2. 买入: +2 个
3. 吃掉: -1 个
4. 最终: 3 + 2 - 1 = 4 个

答案: 4 个

问题: 某商品原价 100 元，先打 8 折，再打 9 折，最终价格是多少？

步骤:
"""

response = llm.generate(prompt)

# 提供示例
prompt = """你是一个代码翻译专家。将 Python 代码翻译为 Go。

Example 1:
Input: print("Hello World")
Output: package main
import "fmt"
func main() {
    fmt.Println("Hello World")
}

Example 2:
Input: x = [i for i in range(10) if i % 2 == 0]
Output: package main
import "fmt"
func main() {
    var x []int
    for i := 0; i < 10; i++ {
        if i % 2 == 0 {
            x = append(x, i)
        }
    }
    fmt.Println(x)
}

现在请翻译:
Input: def add(a, b):
    return a + b
Output:
"""

# 多次采样，投票选择最优
prompts = []
for i in range(5):
    prompt = f"""请解答: {question}
    
    步骤:
    """
    prompts.append(prompt)

responses = [llm.generate(p) for p in prompts]

# 选择最常见的答案
from collections import Counter
answers = [extract_answer(r) for r in responses]
final_answer = Counter(answers).most_common(1)[0][0]

Agent
├── Memory (记忆)
│   ├── Short term (当前对话)
│   └── Long term (持久化知识)
├── Tools (工具)
│   ├── Search
│   ├── Calculator
│   └── File System
├── Planning (规划)
│   ├── Task decomposition
│   └── Execution strategy
└── Execution (执行)
    └── Tool calling

import json
from typing import List, Dict, Any

class Agent:
    def __init__(self, llm, tools: List[BaseTool]):
        self.llm = llm
        self.tools = {t.name: t for t in tools}
        self.memory = []
    
    def run(self, query: str) -> str:
        # 1. 规划任务
        plan = self._plan(query)
        
        # 2. 执行任务
        results = []
        for task in plan:
            result = self._execute(task)
            results.append(result)
        
        # 3. 合成结果
        final = self._synthesize(results)
        
        # 4. 存储记忆
        self.memory.append({"query": query, "response": final})
        
        return final
    
    def _plan(self, query: str) -> List[Dict]:
        prompt = f"""你是一个任务规划器。
        将以下查询分解为可执行的任务。

        Query: {query}

        Available tools: {list(self.tools.keys())}

        Output JSON format:
        {{
            "tasks": [
                {{"tool": "tool_name", "args": {"..."}}},
            ]
        }}
        """
        
        response = self.llm.generate(prompt)
        return json.loads(response)["tasks"]
    
    def _execute(self, task: Dict) -> Any:
        tool_name = task["tool"]
        args = task["args"]
        
        if tool_name not in self.tools:
            return f"Unknown tool: {tool_name}"
        
        tool = self.tools[tool_name]
        return tool.run(**args)
    
    def _synthesize(self, results: List[Any]) -> str:
        prompt = f"""你是一个结果合成器。
        将以下执行结果合成为最终答案。

        Results:
        {json.dumps(results, indent=2)}

        Output: 原生的自然语言答案
        """
        
        return self.llm.generate(prompt)

class AdvancedAgent:
    def __init__(self, llm, tools, memory):
        self.llm = llm
        self.tools = tools
        self.memory = memory
        self.max_iterations = 5
    
    def run(self, query: str) -> str:
        # 系统提示
        system_prompt = self._build_system_prompt()
        
        # 初始化对话
        messages = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": query}
        ]
        
        # 循环执行
        for i in range(self.max_iterations):
            # 调用 LLM
            response = self.llm.generate(messages, tools=self.tools)
            
            # 检查是否需要终止
            if response.get("finish_reason") == "stop":
                return response["content"]
            
            # 处理工具调用
            if response.get("tool_calls"):
                messages.append(response)
                
                for tool_call in response["tool_calls"]:
                    result = self._execute_tool(tool_call)
                    messages.append({
                        "role": "tool",
                        "tool_call_id": tool_call["id"],
                        "content": result
                    })
            
            # 上下文窗口限制
            messages = self._trim_context(messages)
        
        # 达到最大迭代次数
        return "Error: Maximum iterations reached"
    
    def _build_system_prompt(self) -> str:
        return f"""你是一个专业的 AI Agent。
        
        Available tools:
        {self._list_tools()}

        Guidelines:
        1. 分析用户需求
        2. 规划解决方案
        3. 调用工具执行
        4. 合成结果
        5. 保持 helpful 和 honest
        """
    
    def _list_tools(self) -> str:
        return "\n".join([
            f"- {tool.name}: {tool.description}"
            for tool in self.tools
        ])

from langchain import LLMChain, PromptTemplate
from langchain.agents import AgentType, initialize_agent
from langchain.tools import Tool
from langchain.memory import ConversationBufferMemory

# 1. LLM
llm = OpenAI(model_name="gpt-3.5-turbo", temperature=0)

# 2. Tools
tools = [
    Tool(
        name="Search",
        func=lambda q: search_engine.query(q),
        description="用于搜索信息"
    ),
    Tool(
        name="Calculator",
        func=lambda expr: eval(expr),
        description="用于计算数学表达式"
    )
]

# 3. Memory
memory = ConversationBufferMemory(memory_key="chat_history")

# 4. Agent
agent = initialize_agent(
    tools,
    llm,
    agent=AgentType.CONVERSATIONAL_REACT_DESCRIPTION,
    memory=memory,
    verbose=True
)

# 运行
response = agent.run("今天北京天气如何？然后帮我算 123 * 456")

from langchain.chains import LLMChain
from langchain.prompts import ChatPromptTemplate

# 自定义 Prompt
prompt = ChatPromptTemplate.from_template(
    """你是一位资深 Python 开发者。

    请根据以下需求编写代码：
    {requirements}

    要求：
    1. 使用最新 Python 特性
    2. 包含类型注解
    3. 有 docstring
    4. 代码简洁高效
    
    Code:
    """
)

# 创建 Chain
code_chain = LLMChain(
    llm=OpenAI(temperature=0.2),
    prompt=prompt,
    output_key="code"
)

# 连续 Chain
from langchain.chains import SequentialChain

workflow = SequentialChain(
    chains=[code_chain],
    input_variables=["requirements"],
    output_variables=["code"]
)

result = workflow.run("实现一个用户认证系统")

class AgentREPL:
    def __init__(self, agent):
        self.agent = agent
        self.history = []
    
    def run(self):
        print("AI Agent REPL - 输入 'quit' 退出")
        
        while True:
            try:
                user_input = input(">> ").strip()
                
                if user_input.lower() in ['quit', 'exit']:
                    break
                
                if not user_input:
                    continue
                
                # 执行
                response = self.agent.run(user_input)
                print(response)
                
                # 记录历史
                self.history.append((user_input, response))
                
            except KeyboardInterrupt:
                break
            except Exception as e:
                print(f"Error: {e}")

import streamlit as st
from langchain.callbacks import StreamlitCallbackHandler

st.set_page_config(page_title="AI Agent", page_icon="🤖")

# 初始化
if "agent" not in st.session_state:
    st.session_state.agent = initialize_agent(...)

# 显示历史
for msg in st.session_state.history:
    st.chat_message("user").write(msg[0])
    st.chat_message("assistant").write(msg[1])

# 输入
if prompt := st.chat_input("Ask something..."):
    st.chat_message("user").write(prompt)
    
    # Stream handler
    handler = StreamlitCallbackHandler(st.container())
    
    # Run
    response = st.session_state.agent.run(prompt, callbacks=[handler])
    st.session_state.history.append((prompt, response))

from langchain.tools import BaseTool
from pydantic import BaseModel, Field

class WebSearchInput(BaseModel):
    query: str = Field(description="搜索查询")

class WebSearchTool(BaseTool):
    name = "web_search"
    description = "在互联网上搜索信息"
    args_schema: Type[BaseModel] = WebSearchInput
    
    def _run(self, query: str) -> str:
        # 使用百度/Google API
        results = search_engine.search(query)
        
        # 格式化结果
        return "\n".join([
            f"Title: {r.title}\nURL: {r.url}\nSnippet: {r.snippet}"
            for r in results[:5]
        ])
    
    async def _arun(self, query: str) -> str:
        # 异步实现
        ...

import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("agent")

# 日志工具调用
@tool
def search(query: str) -> str:
    logger.info(f"Calling search with: {query}")
    result = do_search(query)
    logger.info(f"Search returned: {len(result)} items")
    return result