腾讯全栈智能体架构：企业级AI入口的战略布局

from tencentcloud.common import credential
from tencentcloud.common.profile.client_profile import ClientProfile
from tencentcloud.common.profile.http_profile import HttpProfile
from tencentcloud.hunyuan.v20230901 import hunyuan_client, models
import json


def hunyuan_chat(messages, tools=None):
    """调用混元大模型API

    Args:
        messages: 对话消息列表
        tools: 可选的工具定义列表

    Returns:
        模型回复或工具调用请求
    """
    cred = credential.Credential("YOUR_SECRET_ID", "YOUR_SECRET_KEY")
    http_profile = HttpProfile()
    http_profile.endpoint = "hunyuan.tencentcloudapi.com"

    client_profile = ClientProfile()
    client_profile.httpProfile = http_profile
    client = hunyuan_client.HunyuanClient(
        cred, "ap-guangzhou", client_profile
    )

    req = models.ChatCompletionsRequest()
    req.Messages = messages

    if tools:
        req.Tools = json.dumps(tools)

    resp = client.ChatCompletions(req)
    result = json.loads(resp.Choices[0].Message.Content)
    return result


# 示例 1：基础文本生成
messages = [
    {"role": "system", "content": "你是一个专业的金融分析师助手。"},
    {"role": "user", "content": "请分析2026年Q1中国AI行业的市场趋势。"}
]
response = hunyuan_chat(messages)
print(response)


# 示例 2：工具调用
search_tool = {
    "type": "function",
    "function": {
        "name": "search_knowledge_base",
        "description": "在企业知识库中搜索相关文档",
        "parameters": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "搜索关键词"
                },
                "category": {
                    "type": "string",
                    "enum": ["技术文档", "产品手册", "合规文件"]
                }
            },
            "required": ["query"]
        }
    }
}

messages = [
    {"role": "user", "content": "我们的数据安全合规要求有哪些？"}
]
response = hunyuan_chat(messages, tools=[search_tool])

if "tool_calls" in response:
    for tool_call in response["tool_calls"]:
        func_name = tool_call["function"]["name"]
        func_args = json.loads(tool_call["function"]["arguments"])
        print(f"调用工具: {func_name}, 参数: {func_args}")

from tencent_agent import Agent, Tool, Memory, PlanStrategy


# 定义工具：CRM系统集成
class CRMTool(Tool):
    name = "crm_integration"
    description = "查询和管理客户关系数据"

    def execute(self, customer_id, action="query"):
        if action == "query":
            return self._query_customer(customer_id)
        elif action == "update":
            return self._update_customer(customer_id)
        else:
            raise ValueError(f"Unknown action: {action}")

    def _query_customer(self, customer_id):
        return {
            "name": "张三",
            "tier": "VIP",
            "history": ["订单A", "订单B"]
        }


# 定义工具：知识库检索
class KnowledgeBaseTool(Tool):
    name = "knowledge_search"
    description = "在企业知识库中搜索产品信息和FAQ"

    def execute(self, query, top_k=3):
        results = self._vector_search(query, top_k)
        return results


# 定义记忆策略
customer_memory = Memory(
    name="customer_profile",
    storage="vector+structured",
    ttl=86400 * 30,
    retrieval_strategy="semantic+filter"
)


# 构建智能客服Agent
customer_service_agent = Agent(
    name="智能客服助手",
    model="hunyuan-pro",
    system_prompt="""你是一个专业的企业智能客服助手。
你的职责是：
1. 准确理解客户问题
2. 查询CRM系统获取客户信息
3. 搜索知识库找到相关解答
4. 给出清晰、专业的回复
5. 记录重要交互到客户档案""",
    tools=[CRMTool(), KnowledgeBaseTool()],
    memory=customer_memory,
    plan_strategy=PlanStrategy.REACT,
    max_iterations=5
)


# 运行Agent
def handle_customer_query(customer_id, query):
    result = customer_service_agent.run(
        user_input=query,
        context={"customer_id": customer_id}
    )
    return {
        "response": result.output,
        "tool_calls": result.tool_calls,
        "memory_updated": result.memory_ops
    }

result = handle_customer_query(
    customer_id="CUST-20260610-001",
    query="我的订单什么时候发货？"
)
print(result["response"])

// 小程序端前端
Page({
  data: {
    messages: [],
    inputText: '',
    isLoading: false
  },

  async sendMessage() {
    const text = this.data.inputText.trim();
    if (!text || this.data.isLoading) return;

    const userMsg = { role: 'user', content: text };
    this.setData({
      messages: [...this.data.messages, userMsg],
      inputText: '',
      isLoading: true
    });

    try {
      const res = await wx.cloud.callFunction({
        name: 'ai-service',
        data: {
          action: 'chat',
          messages: [...this.data.messages, userMsg],
          sessionId: this.getSessionId()
        }
      });

      const aiMsg = {
        role: 'assistant',
        content: res.result.content
      };
      this.setData({
        messages: [...this.data.messages, aiMsg],
        isLoading: false
      });
    } catch (err) {
      this.setData({ isLoading: false });
      wx.showToast({
        title: '服务暂时不可用',
        icon: 'none'
      });
    }
  },

  getSessionId() {
    let sid = wx.getStorageSync('sessionId');
    if (!sid) {
      sid = 'sess_' + Date.now();
      wx.setStorageSync('sessionId', sid);
    }
    return sid;
  }
});


// 云函数端后端
const cloud = require('wx-server-sdk');
cloud.init({ env: cloud.DYNAMIC_CURRENT_ENV });

exports.main = async (event) => {
  const { action, messages, sessionId } = event;

  if (action === 'chat') {
    return await handleChat(messages, sessionId);
  }
};

async function handleChat(messages, sessionId) {
  // 1. 内容安全审核
  const lastMsg = messages[messages.length - 1].content;
  const inputSafe = await checkContentSafety(lastMsg);
  if (!inputSafe) {
    return { content: '输入包含不当内容，请重新输入。' };
  }

  // 2. 检查缓存
  const cacheKey = `chat:${sessionId}:${lastMsg}`;
  const cached = await getFromCache(cacheKey);
  if (cached) {
    return { content: cached, source: 'cache' };
  }

  // 3. 调用混元大模型
  const systemMsg = {
    role: 'system',
    content: '你是企业智能客服助手，请简洁准确回答。'
  };
  const fullMessages = [systemMsg, ...messages.slice(-10)];
  const result = await callHunyuan(fullMessages);

  // 4. 输出审核 + 缓存 + 持久化
  const outputSafe = await checkContentSafety(result);
  if (!outputSafe) {
    return { content: '抱歉，暂时无法回答。' };
  }

  await saveToCache(cacheKey, result, 300);
  await saveToDB(sessionId, messages, result);

  return { content: result };
}

from flask import Flask, request, jsonify
from tencent_agent import Agent, Tool

app = Flask(__name__)


# 工具：发票验证
class InvoiceVerifyTool(Tool):
    name = "invoice_verify"
    description = "验证发票真伪及金额"

    def execute(self, invoice_code, invoice_number, amount):
        result = self._call_tax_api(
            invoice_code, invoice_number, amount
        )
        return {
            "is_valid": result["valid"],
            "actual_amount": result["amount"],
            "issuer": result["issuer"]
        }


# 工具：预算检查
class BudgetCheckTool(Tool):
    name = "budget_check"
    description = "检查部门预算余额"

    def execute(self, department, category):
        budget = self._get_budget(department, category)
        return {
            "total": budget["total"],
            "used": budget["used"],
            "remaining": budget["remaining"],
            "within_budget": budget["remaining"] > 0
        }


# 工具：异常检测
class AnomalyDetectTool(Tool):
    name = "anomaly_detect"
    description = "检测报销异常模式"

    def execute(self, user_id, amount, category):
        avg = self._get_user_avg(user_id, category)
        score = abs(amount - avg) / max(avg, 1)
        return {
            "is_anomaly": score > 2.0,
            "score": round(score, 2),
            "alert": "金额显著高于历史平均" if score > 2.0 else "正常"
        }


# 构建报销审批Agent
expense_agent = Agent(
    name="报销审批助手",
    model="hunyuan-pro",
    system_prompt="""你是企业报销审批助手。流程：
1. 接收报销申请
2. 验证发票真伪
3. 检查部门预算余额
4. 检测报销异常模式
5. 给出审批建议：通过/驳回/转人工

必须基于事实和数据做判断。""",
    tools=[InvoiceVerifyTool(), BudgetCheckTool(), AnomalyDetectTool()],
    max_iterations=6
)


@app.route("/wecom/callback", methods=["POST"])
def wecom_callback():
    data = request.json
    user_id = data.get("UserId")
    message = data.get("Content")

    result = expense_agent.run(
        user_input=message,
        context={"user_id": user_id}
    )

    return jsonify({
        "touser": user_id,
        "msgtype": "text",
        "text": {"content": result.output}
    })


if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8080)