# 使用UV安装依赖uv add openai pydantic-ai python-dotenv
# 设置环境变量echo "OPENAI_API_KEY=your-api-key-here" > .env
# 运行基础示例uv run basic_intelligence.py

# 克隆项目git clone <repository-url>cd ai-agents-guide
# 同步所有依赖uv sync
# 安装开发依赖uv add --dev pytest black ruff

from openai import OpenAI
def basic_intelligence(prompt: str) -> str:    client = OpenAI()    response = client.responses.create(model="gpt-4o", input=prompt)    return response.output_text
# 使用示例result = basic_intelligence("什么是人工智能？")print(result)

from pydantic import BaseModelfrom openai import OpenAI
class TaskResult(BaseModel):    task: str    completed: bool    priority: int
def structured_intelligence(prompt: str) -> TaskResult:    client = OpenAI()    response = client.responses.parse(        model="gpt-4o",        input=[            {"role": "system", "content": "从用户输入中提取任务信息"},            {"role": "user", "content": prompt},        ],        text_format=TaskResult,    )    return response.output_parsed
# 使用示例result = structured_intelligence("我需要在周五前完成项目演示，这是高优先级任务")print(f"提取的任务: {result.task}")

import requestsfrom openai import OpenAI
def get_weather(latitude, longitude):    response = requests.get(        f"https://api.open-meteo.com/v1/forecast?latitude={latitude}&longitude={longitude}&current=temperature_2m"    )    data = response.json()    return data["current"]["temperature_2m"]
def intelligence_with_tools(prompt: str) -> str:    client = OpenAI()        tools = [        {            "type": "function",            "name": "get_weather",            "description": "获取指定坐标的当前温度（摄氏度）",            "parameters": {                "type": "object",                "properties": {                    "latitude": {"type": "number"},                    "longitude": {"type": "number"},                },                "required": ["latitude", "longitude"],            }        }    ]        # 模型调用和工具执行逻辑    # ...    return final_response
# 使用示例result = intelligence_with_tools("今天巴黎的天气怎么样？")print(result)

from typing import Literalfrom pydantic import BaseModelfrom openai import OpenAI
class IntentClassification(BaseModel):    intent: Literal["question", "request", "complaint"]    confidence: float    reasoning: str
def route_based_on_intent(user_input: str) -> tuple[str, IntentClassification]:    client = OpenAI()    response = client.responses.parse(        model="gpt-4o",        input=[            {                "role": "system",                 "content": "将用户输入分类为：问题、请求或投诉"            },            {"role": "user", "content": user_input},        ],        text_format=IntentClassification,    )
    classification = response.output_parsed    intent = classification.intent
    # 基于意图路由到不同处理函数    if intent == "question":        result = answer_question(user_input)    elif intent == "request":        result = process_request(user_input)    elif intent == "complaint":        result = handle_complaint(user_input)    else:        result = "我不确定如何帮助您"
    return result, classification

from mem0 import Memoryfrom dotenv import load_dotenv
load_dotenv()
class CustomerSupportAIAgent:    def __init__(self):        config = {            "vector_store": {                "provider": "qdrant",                "config": {"host": "localhost", "port": 6333},            },        }        self.memory = Memory.from_config(config)        self.client = OpenAI()
    def handle_query(self, query, user_id=None):        # 检索相关记忆        relevant_memories = self.memory.search(query=query, user_id=user_id, limit=3)                # 生成响应并存储新记忆        self.memory.add(query, user_id=user_id, metadata={"app_id": "customer-support"})                return response

import asynciofrom openai import AsyncOpenAIfrom pydantic import BaseModel, Field
class CalendarValidation(BaseModel):    is_calendar_request: bool    confidence_score: float
class SecurityCheck(BaseModel):    is_safe: bool    risk_flags: list[str]
async def validate_request(user_input: str) -> bool:    """并行运行验证检查"""    calendar_check, security_check = await asyncio.gather(        validate_calendar_request(user_input),        check_security(user_input)    )
    return (        calendar_check.is_calendar_request and        calendar_check.confidence_score > 0.7 and        security_check.is_safe    )

from mcp.server.fastmcp import FastMCP
# 创建MCP服务器mcp = FastMCP("KnowledgeBase")
@mcp.tool()def get_knowledge_base() -> str:    """检索整个知识库"""    try:        with open("data/kb.json", "r") as f:            kb_data = json.load(f)        return format_knowledge_base(kb_data)    except Exception as e:        return f"错误: {str(e)}"
# 运行服务器if __name__ == "__main__":    mcp.run(transport="stdio")

from docling.document_converter import DocumentConverterfrom docling.chunking import HybridChunkerimport lancedb
def process_document(url: str):    # 文档转换    converter = DocumentConverter()    result = converter.convert(url)        # 智能分块    chunker = HybridChunker(max_tokens=8191)    chunks = list(chunker.chunk(result.document))        # 向量存储    db = lancedb.connect("data/lancedb")    table = db.create_table("documents", schema=ChunkSchema)    table.add(processed_chunks)        return chunks