AI Agent 开发

# LLM 推理示例 - 使用 OpenAI API
from openai import OpenAI

client = OpenAI(api_key="your-api-key")

response = client.chat.completions.create(
    model="gpt-4",
    messages=[
        {"role": "system", "content": "你是一个专业的编程助手。"},
        {"role": "user", "content": "解释一下什么是装饰器"}
    ],
    temperature=0.7,
    max_tokens=500,
    top_p=0.9
)

print(response.choices[0].message.content)

# API 调用示例 - 多模型统一接口
import openai

# OpenAI
openai_client = openai.OpenAI(api_key="sk-xxx")

# 兼容 OpenAI 接口的其他模型（如通义千问、DeepSeek）
qwen_client = openai.OpenAI(
    api_key="sk-xxx",
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)

# vLLM 部署示例 - 使用 vLLM Python API
from vllm import LLM, SamplingParams

llm = LLM(model="meta-llama/Llama-3-8b")
sampling_params = SamplingParams(temperature=0.8, top_p=0.95)

outputs = llm.generate(["Hello, my name is"], sampling_params)
print(outputs[0].outputs[0].text)

# 好的 Prompt 示例
prompt = """
你是一个资深的技术架构师。请分析以下系统架构方案，并按照以下格式输出：

【方案概述】用一句话总结方案核心
【优点】列出 3 个主要优势
【风险】列出 2 个潜在风险
【建议】给出 1 条改进建议

待分析方案：{architecture_description}
"""

# Few-shot Prompt 示例
prompt = """
根据以下示例，提取句子中的关键信息：

输入：张三昨天在北京的会议上介绍了新的机器学习算法
输出：{"人物": "张三", "时间": "昨天", "地点": "北京", "事件": "介绍新机器学习算法"}

输入：李四团队上周完成了系统性能优化，响应时间降低了 50%
输出：{"人物": "李四团队", "时间": "上周", "地点": "未提及", "事件": "完成系统性能优化"}

输入：{user_input}
输出：
"""

from string import Template

# 简单模板示例
code_review_template = Template("""
你是一个资深代码审查员。请审查以下 Python 代码：

```python
$code

# RAG 基础流程示例 - 使用 LangChain
from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_chroma import Chroma
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import create_retrieval_chain

# 1. 加载文档
loader = TextLoader("knowledge_base.txt")
documents = loader.load()

# 2. 分块
text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
chunks = text_splitter.split_documents(documents)

# 3. 向量化并存储
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(chunks, embeddings)

# 4. 检索
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

# 5. 生成
prompt = ChatPromptTemplate.from_template("""
基于以下上下文回答问题：
{context}

问题：{input}
""")

llm = ChatOpenAI(model="gpt-4")
retrieval_chain = create_retrieval_chain(retriever, prompt | llm)

response = retrieval_chain.invoke({"input": "什么是 RAG？"})
print(response["answer"])

from langchain_text_splitters import (
    RecursiveCharacterTextSplitter,
    MarkdownHeaderTextSplitter,
    Language
)

# 递归字符分割
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    length_function=len,
    separators=["\n\n", "\n", "。", "！", "？", " ", ""]
)

# Markdown 分割
headers_to_split_on = [
    ("#", "Header 1"),
    ("##", "Header 2"),
    ("###", "Header 3"),
]
markdown_splitter = MarkdownHeaderTextSplitter(headers_to_split_on)

# 代码分割
python_splitter = RecursiveCharacterTextSplitter.from_language(
    language=Language.PYTHON,
    chunk_size=500,
    chunk_overlap=100
)

# 使用不同 Embedding 模型
from langchain_openai import OpenAIEmbeddings
from langchain_community.embeddings import HuggingFaceBgeEmbeddings

# OpenAI Embedding
openai_embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small",
    dimensions=1536
)

# BGE Embedding（本地部署）
bge_embeddings = HuggingFaceBgeEmbeddings(
    model_name="BAAI/bge-large-zh-v1.5",
    model_kwargs={'device': 'cuda'},
    encode_kwargs={'normalize_embeddings': True}
)

# 计算文本向量
text = "人工智能是未来的发展方向"
vector = openai_embeddings.embed_query(text)
print(f"向量维度：{len(vector)}")

# 混合检索示例
from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever
from langchain_chroma import Chroma

# 向量检索器
vector_retriever = Chroma.from_documents(docs, embeddings).as_retriever(
    search_kwargs={"k": 5}
)

# 关键词检索器
bm25_retriever = BM25Retriever.from_documents(docs)
bm25_retriever.k = 5

# 混合检索
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.4, 0.6]  # 权重分配
)

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

# 构建简单链
prompt = ChatPromptTemplate.from_template("请解释以下概念：{concept}")
llm = ChatOpenAI(model="gpt-4")
output_parser = StrOutputParser()

# 使用 LCEL（LangChain Expression Language）组合
chain = prompt | llm | output_parser

result = chain.invoke({"concept": "微服务架构"})
print(result)

from langchain_core.prompts import (
    ChatPromptTemplate,
    SystemMessagePromptTemplate,
    HumanMessagePromptTemplate
)

# 对话模板
chat_prompt = ChatPromptTemplate.from_messages([
    SystemMessagePromptTemplate.from_template("你是{role}专家"),
    HumanMessagePromptTemplate.from_template("请分析：{question}")
])

messages = chat_prompt.format_messages(role="安全", question="SQL注入防护")

from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain

# 对话记忆
memory = ConversationBufferMemory()
conversation = ConversationChain(
    llm=ChatOpenAI(temperature=0.7),
    memory=memory,
    verbose=True
)

# 多轮对话
conversation.run("你好，我想学习 Python")
conversation.run("推荐一些学习资源")

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages

# 定义状态
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    task_status: str
    result: str

# 定义节点函数
def research_node(state: AgentState):
    return {"task_status": "researching", "result": "研究结果..."}

def review_node(state: AgentState):
    return {"task_status": "reviewed", "result": state["result"] + " [已审查]"}

# 构建图
workflow = StateGraph(AgentState)
workflow.add_node("research", research_node)
workflow.add_node("review", review_node)

workflow.set_entry_point("research")
workflow.add_edge("research", "review")
workflow.add_edge("review", END)

app = workflow.compile()

def should_continue(state: AgentState) -> str:
    if "error" in state.get("result", ""):
        return "error_handler"
    return "finalizer"

workflow.add_conditional_edges("review", should_continue)

import os

# 配置 LangSmith
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-api-key"
os.environ["LANGCHAIN_PROJECT"] = "my-agent-project"

from langchain.agents import create_openai_functions_agent, AgentExecutor
from langchain.tools import DuckDuckGoSearchRun, Tool

# 定义工具
search = DuckDuckGoSearchRun()
calculator = Tool(
    name="Calculator",
    func=lambda x: str(eval(x)),
    description="用于数学计算"
)

tools = [search, calculator]

# 创建 Agent
agent = create_openai_functions_agent(
    llm=ChatOpenAI(model="gpt-4"),
    tools=tools,
    prompt=ChatPromptTemplate.from_messages([
        ("system", "你是一个有帮助的助手，可以使用工具回答问题。"),
        ("human", "{input}")
    ])
)

agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
result = agent_executor.invoke({"input": "今天北京天气如何？123 * 456 = ?"})

from langchain.tools import BaseTool

class DatabaseQueryTool(BaseTool):
    name: str = "db_query"
    description: str = "查询数据库，参数：SQL 查询语句"
    
    def _run(self, query: str) -> str:
        # 执行数据库查询
        return execute_sql(query)

# ReAct Agent 示例
from langchain.agents import initialize_agent, AgentType
from langchain.tools import Tool

tools = [
    Tool(
        name="Search",
        func=search_tool,
        description="用于搜索网络信息"
    ),
    Tool(
        name="Calculator",
        func=calculator_tool,
        description="用于数学计算"
    )
]

agent = initialize_agent(
    tools=tools,
    llm=ChatOpenAI(model="gpt-4"),
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True
)

# Agent 会交替进行思考和行动
# Thought: 我需要先搜索相关信息
# Action: Search[查询内容]
# Observation: 搜索结果...
# Thought: 根据结果我需要计算...
# Action: Calculator[计算公式]
# Observation: 计算结果...
# Thought: 现在我可以回答问题了
# Final Answer: 最终回答

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_openai import ChatOpenAI

# 规划器
class Plan(BaseModel):
    steps: list[str] = Field(description="任务步骤列表")

planner_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个任务规划专家。将复杂任务分解为可执行的步骤。"),
    ("human", "任务：{task}")
])

llm = ChatOpenAI(model="gpt-4")
planner = planner_prompt | llm.with_structured_output(Plan)

# 执行规划
plan = planner.invoke({"task": "分析竞品产品并生成对比报告"})
print(f"执行计划：{plan.steps}")

# 执行器依次处理每个步骤
for i, step in enumerate(plan.steps):
    print(f"执行步骤 {i+1}: {step}")
    # result = executor.invoke({"step": step})

from crewai import Agent, Task, Crew

# 定义角色
researcher = Agent(
    role='高级研究员',
    goal='深入研究指定技术领域',
    backstory='你是一位有 10 年经验的技术研究员',
    verbose=True,
    allow_delegation=False
)

writer = Agent(
    role='技术写作者',
    goal='将研究成果整理为结构化的技术文档',
    backstory='你是一位擅长将复杂技术概念通俗化的写作者',
    verbose=True,
    allow_delegation=False
)

# 定义任务
research_task = Task(
    description='研究 AI Agent 的最新发展趋势',
    agent=researcher,
    expected_output='研究报告'
)

writing_task = Task(
    description='根据研究报告撰写技术文档',
    agent=writer,
    expected_output='技术文档',
    context=[research_task]  # 依赖前一个任务的结果
)

# 组建团队
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
    verbose=True
)

result = crew.kickoff()

from langchain.memory import (
    ConversationBufferMemory,
    ConversationSummaryMemory,
    ConversationBufferWindowMemory
)

# 滑动窗口记忆 - 只保留最近 5 轮
window_memory = ConversationBufferWindowMemory(k=5)

# 摘要记忆 - 使用 LLM 总结历史对话
summary_memory = ConversationSummaryMemory(
    llm=ChatOpenAI(temperature=0),
    max_token_limit=2000
)

# 工作记忆示例
class TaskMemory:
    def __init__(self):
        self.current_task = None
        self.subtasks = []
        self.completed = []
        self.intermediate_results = {}
    
    def add_subtask(self, subtask: str):
        self.subtasks.append(subtask)
    
    def complete_subtask(self, index: int, result: str):
        task = self.subtasks.pop(index)
        self.completed.append(task)
        self.intermediate_results[task] = result

# Milvus 使用示例
from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType

# 连接 Milvus
connections.connect("default", host="localhost", port="19530")

# 定义 Collection Schema
fields = [
    FieldSchema(name="id", dtype=DataType.INT64, is_primary=True),
    FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=1536),
    FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=2000)
]
schema = CollectionSchema(fields, description="文档向量库")

# 创建 Collection
collection = Collection("documents", schema)

# 插入数据
entities = [
    [1, 2],  # IDs
    [[0.1] * 1536, [0.2] * 1536],  # Embeddings
    ["文本1", "文本2"]  # Texts
]
collection.insert(entities)

# 创建索引
index_params = {
    "index_type": "HNSW",
    "metric_type": "COSINE",
    "params": {"M": 16, "efConstruction": 64}
}
collection.create_index("embedding", index_params)

# 搜索
collection.load()
results = collection.search(
    data=[[0.15] * 1536],
    anns_field="embedding",
    param={"metric_type": "COSINE", "params": {"ef": 64}},
    limit=3
)

# Chroma 使用示例
import chromadb
from chromadb.utils import embedding_functions

# 创建客户端（内存模式）
client = chromadb.Client()

# 或持久化模式
# client = chromadb.PersistentClient(path="./chroma_db")

# 创建 Collection
collection = client.create_collection(name="documents")

# 添加文档
collection.add(
    documents=["这是第一篇文章", "这是第二篇文章"],
    metadatas=[{"source": "web"}, {"source": "file"}],
    ids=["id1", "id2"]
)

# 查询
results = collection.query(
    query_texts=["关于什么？"],
    n_results=2
)

# Pinecone 使用示例
import pinecone

# 初始化
pinecone.init(api_key="your-api-key", environment="us-east-1")

# 创建索引
pinecone.create_index(
    name="my-index",
    dimension=1536,
    metric="cosine"
)

# 连接索引
index = pinecone.Index("my-index")

# 插入数据
index.upsert([
    ("id1", [0.1] * 1536, {"text": "内容1"}),
    ("id2", [0.2] * 1536, {"text": "内容2"})
])

# 查询
results = index.query(
    vector=[0.15] * 1536,
    top_k=3,
    include_metadata=True
)

import time
from functools import wraps
from openai import RateLimitError, APIError

def retry_with_backoff(max_retries=3, base_delay=1):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except (RateLimitError, APIError) as e:
                    if attempt == max_retries - 1:
                        raise
                    delay = base_delay * (2 ** attempt)
                    time.sleep(delay)
            return None
        return wrapper
    return decorator

@retry_with_backoff(max_retries=3)
def call_llm(prompt):
    return client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": prompt}]
    )

# 降级策略示例
def get_response_with_fallback(prompt):
    # 尝试主模型
    try:
        return call_gpt4(prompt)
    except Exception:
        # 降级到 GPT-3.5
        try:
            return call_gpt35(prompt)
        except Exception:
            # 返回缓存结果
            return get_cached_response(prompt)

# 语义缓存示例
from langchain.globals import set_llm_cache
from langchain.cache import SQLiteCache

# 启用缓存
set_llm_cache(SQLiteCache(database_path=".langchain.db"))

# 相同请求会命中缓存
response1 = llm.invoke("什么是 Python？")
response2 = llm.invoke("什么是 Python？")  # 从缓存返回

# 模型路由示例
def route_model(task_complexity: str, prompt: str):
    if task_complexity == "simple":
        return call_small_model(prompt)  # 如 GPT-3.5
    elif task_complexity == "complex":
        return call_large_model(prompt)  # 如 GPT-4
    else:
        return call_medium_model(prompt)  # 如 GPT-4o-mini

# Prompt 注入防护示例
def sanitize_user_input(user_input: str) -> str:
    # 移除潜在的注入指令
    dangerous_patterns = [
        "ignore previous instructions",
        "system:",
        "you are now",
        "disregard"
    ]
    for pattern in dangerous_patterns:
        if pattern.lower() in user_input.lower():
            raise ValueError("Potential prompt injection detected")
    return user_input