tx_flow_analysis/examples/basic_agent.py

#!/usr/bin/env python3
"""
基础Agent示例 - LangChain学习起点
=================================

这个文件展示了如何创建一个简单的Agent，包含：
1. 基础的LLM调用
2. 简单的提示词设计
3. 基本的结果处理

运行方法：
python examples/basic_agent.py
"""

import os
import sys
from typing import Dict, Any
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()

try:
    from langchain_openai import ChatOpenAI
    from langchain_core.prompts import ChatPromptTemplate
except ImportError as e:
    print(f"❌ 缺少依赖包: {e}")
    print("请运行: pip install langchain langchain-openai python-dotenv")
    sys.exit(1)


class BasicAgent:
    """基础Agent示例"""

    def __init__(self):
        """初始化Agent"""
        api_key = os.getenv('DEEPSEEK_API_KEY')
        if not api_key:
            raise ValueError("请在.env文件中设置DEEPSEEK_API_KEY")

        # 初始化LLM
        self.llm = ChatOpenAI(
            model="deepseek-chat",
            api_key=api_key,
            base_url="https://api.deepseek.com",
            temperature=0.1
        )

        # 创建提示词模板
        self.prompt_template = ChatPromptTemplate.from_messages([
            ("system", "你是一个专业的数据分析师，请简洁地回答用户的问题。"),
            ("user", "{question}")
        ])

        print("✅ BasicAgent初始化完成")

    def analyze_data(self, question: str, data_sample: str = None) -> Dict[str, Any]:
        """
        分析数据并返回结果

        Args:
            question: 用户问题
            data_sample: 数据样本（可选）

        Returns:
            分析结果字典
        """
        try:
            # 构建完整的查询
            full_question = question
            if data_sample:
                full_question += f"\n\n数据样本: {data_sample}"

            # 创建链并调用
            chain = self.prompt_template | self.llm
            response = chain.invoke({"question": full_question})

            return {
                "success": True,
                "question": question,
                "data_sample": data_sample,
                "answer": response.content,
                "model": "deepseek-chat"
            }

        except Exception as e:
            return {
                "success": False,
                "question": question,
                "error": str(e)
            }


def main():
    """主函数 - 演示基本Agent功能"""
    print("🚀 基础Agent示例")
    print("=" * 50)

    try:
        # 创建Agent实例
        agent = BasicAgent()

        # 示例问题
        questions = [
            "请解释什么是数据分析",
            "如何提高数据分析的准确性？",
            "数据可视化的重要性是什么？"
        ]

        print("\n🧪 测试Agent功能:")
        print("-" * 30)

        # 逐个测试问题
        for i, question in enumerate(questions, 1):
            print(f"\n📋 问题 {i}: {question}")

            result = agent.analyze_data(question)

            if result["success"]:
                print(f"✅ 回答: {result['answer'][:100]}...")
            else:
                print(f"❌ 错误: {result['error']}")

        print("\n🎉 基础Agent示例完成！")
        print("\n💡 下一步学习建议:")
        print("1. 查看 examples/state_machine.py - 学习LangGraph状态机")
        print("2. 查看 examples/advanced_agent.py - 学习高级Agent功能")
        print("3. 阅读 PRACTICE_GUIDE.md - 获取完整学习路径")

    except Exception as e:
        print(f"❌ 运行出错: {e}")
        print("\n🔧 故障排除:")
        print("1. 检查 .env 文件是否存在")
        print("2. 确认 DEEPSEEK_API_KEY 已正确设置")
        print("3. 确认网络连接正常")
        print("4. 运行: pip install -r requirements.txt")


if __name__ == "__main__":
    main()