asyncio 无法透传 traceid

Agent内部执行工具时，若使用asyncio线程池执行，则无法透传traceid，即会采集到多条trace，而非一条trace多个span

测试类如下：
```
import asyncio
import concurrent.futures
import logging
import os
import json
import time
from openai import OpenAI

# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s - Thread:%(thread)d',
handlers=[
logging.StreamHandler(),
logging.FileHandler('app.log')
]
)

logger = logging.getLogger(__name__)

# 设置环境变量
os.environ["OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT"] = "true"

class ChatRequest:
def __init__(self, user_id: str, conversation_id: str, message: str):
self.user_id = user_id
self.conversation_id = conversation_id
self.message = message

class ChatResponse:
def __init__(self, content: str):
self.content = content

class ParallelToolsService:
"""并行工具调用服务 - 使用线程池并发执行工具调用"""

def __init__(self):
self.executor = concurrent.futures.ThreadPoolExecutor(max_workers=5)

def parallel_function_call_chat(self, request: ChatRequest) -> ChatResponse:

logger.info(f"开始并行工具调用聊天: {request.conversation_id}")

client = OpenAI()

tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "获取指定城市的天气信息",
"parameters": {
"type": "object",
"properties": {
"city": {"type": "string", "description": "城市名称"},
"date": {"type": "string", "description": "日期，格式：YYYY-MM-DD"}
},
"required": ["city"]
}
}
},
{
"type": "function",
"function": {
"name": "get_stock_price",
"description": "获取股票价格信息",
"parameters": {
"type": "object",
"properties": {
"symbol": {"type": "string", "description": "股票代码"},
"market": {"type": "string", "description": "市场，如：US, CN"}
},
"required": ["symbol"]
}
}
},
{
"type": "function",
"function": {
"name": "search_news",
"description": "搜索相关新闻",
"parameters": {
"type": "object",
"properties": {
"keyword": {"type": "string", "description": "搜索关键词"},
"days": {"type": "integer", "description": "最近几天的新闻"}
},
"required": ["keyword"]
}
}
},
{
"type": "function",
"function": {
"name": "get_exchange_rate",
"description": "获取汇率信息",
"parameters": {
"type": "object",
"properties": {
"from_currency": {"type": "string", "description": "源货币"},
"to_currency": {"type": "string", "description": "目标货币"}
},
"required": ["from_currency", "to_currency"]
}
}
}
]

messages = [{"role": "user", "content": request.message}]

response = client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
tools=tools,
tool_choice="auto",
user=request.user_id
)

assistant_message = response.choices[0].message
messages.append(assistant_message)

if assistant_message.tool_calls:
logger.info(f"检测到 {len(assistant_message.tool_calls)} 个工具调用，开始并发执行")

tool_results = self._execute_tools_concurrently(assistant_message.tool_calls)

# 将所有工具结果添加到消息中
for tool_call, result in zip(assistant_message.tool_calls, tool_results):
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": result
})

logger.info("所有工具调用完成，请求AI总结")

final_response = client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
user=request.user_id
)

return ChatResponse(final_response.choices[0].message.content)
else:
return ChatResponse(assistant_message.content)

def _execute_tools_concurrently(self, tool_calls):

logger.info("使用asyncio和线程池并发执行工具调用")

try:
loop = asyncio.get_event_loop()
except RuntimeError:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)

if loop.is_running():
import concurrent.futures
future = concurrent.futures.Future()

async def run_async():
result = await self._async_execute_tools(tool_calls)
future.set_result(result)

asyncio.create_task(run_async())
return future.result()
else:
return loop.run_until_complete(self._async_execute_tools(tool_calls))

async def _async_execute_tools(self, tool_calls):

logger.info(f"开始异步执行 {len(tool_calls)} 个工具")

loop = asyncio.get_event_loop()

futures = []
for tool_call in tool_calls:
future = loop.run_in_executor(
self.executor,
self._execute_single_tool_in_thread,
tool_call
)
futures.append(future)

logger.info("等待所有工具调用完成...")
results = await asyncio.gather(*futures, return_exceptions=True)

processed_results = []
for i, result in enumerate(results):
if isinstance(result, Exception):
logger.error(f"工具调用 {i} 失败: {result}")
processed_results.append(f"工具调用失败: {str(result)}")
else:
processed_results.append(result)

logger.info(f"所有工具调用完成，成功 {len([r for r in processed_results if not r.startswith('工具调用失败')])} 个")
return processed_results

def _execute_single_tool_in_thread(self, tool_call):

import threading
thread_id = threading.current_thread().ident

tool_name = tool_call.function.name
tool_args = json.loads(tool_call.function.arguments)

logger.info(f"线程 {thread_id} 执行工具: {tool_name}, 参数: {tool_args}")

try:
execution_time = {
"get_weather": 0.5,
"get_stock_price": 0.8,
"search_news": 1.2,
"get_exchange_rate": 0.3
}.get(tool_name, 0.5)

time.sleep(execution_time) # 模拟工具执行时间

result = self._execute_tool_logic(tool_name, tool_args)

logger.info(f"线程 {thread_id} 工具 {tool_name} 执行成功")
return f"[线程{thread_id}] {result}"

except Exception as e:
logger.error(f"线程 {thread_id} 工具 {tool_name} 执行失败: {e}")
return f"[线程{thread_id}] 工具执行失败: {str(e)}"

def _execute_tool_logic(self, tool_name: str, tool_args: dict) -> str:

if tool_name == "get_weather":
city = tool_args.get('city', '未知城市')
date = tool_args.get('date', '今天')
return f"天气查询结果：{city} {date} 晴天，温度25°C，湿度60%，微风"

elif tool_name == "get_stock_price":
symbol = tool_args.get('symbol', '未知股票')
market = tool_args.get('market', 'US')
return f"股票查询结果：{market}市场 {symbol} 当前价格 $150.25 (+2.15% ↗)"

elif tool_name == "search_news":
keyword = tool_args.get('keyword', '')
days = tool_args.get('days', 7)
return f"新闻搜索结果：最近{days}天关于'{keyword}'的热门新闻：1.相关政策发布 2.市场分析报告 3.专家观点解读"

elif tool_name == "get_exchange_rate":
from_curr = tool_args.get('from_currency', 'USD')
to_curr = tool_args.get('to_currency', 'CNY')
return f"汇率查询结果：{from_curr} 到 {to_curr} 当前汇率：7.2385 (更新时间: {time.strftime('%H:%M:%S')})"

return f"未知工具调用: {tool_name}"

def cleanup(self):
"""清理线程池资源"""
logger.info("清理线程池资源")
self.executor.shutdown(wait=True)

def main():
service = ParallelToolsService()

try:
request = ChatRequest(
"user123",
"conv123",
"我想了解一下：1)北京明天的天气如何？2)阿里巴巴(BABA)在美股的最新价格？3)最近一周关于人工智能的新闻？4)美元兑人民币的汇率？请帮我查询这些信息并给出综合分析。"
)

start_time = time.time()
response = service.parallel_function_call_chat(request)
end_time = time.time()

print(f"AI回复: {response.content}")
print(f"\n执行时间: {end_time - start_time:.2f}秒")
print("\n请检查日志文件 parallel_tools_test.log 查看详细的线程执行和trace信息")

except Exception as e:
print(f"测试失败: {e}")
import traceback
traceback.print_exc()

finally:
service.cleanup()

if __name__ == "__main__":
main()
```

包版本如下：